// Copyright (c) 2015-2021 MinIO, Inc.//// This file is part of MinIO Object Storage stack//// This program is free software: you can redistribute it and/or modify// it under the terms of the GNU Affero General Public License as published by// the Free Software Foundation, either version 3 of the License, or// (at your option) any later version.//// This program is distributed in the hope that it will be useful// but WITHOUT ANY WARRANTY; without even the implied warranty of// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the// GNU Affero General Public License for more details.//// You should have received a copy of the GNU Affero General Public License// along with this program.  If not, see <http://www.gnu.org/licenses/>. package cmd import (	"context"	"encoding/binary"	"errors"	"fmt"	"io"	"math"	"net/http"	"path"	"reflect"	"strings"	"sync"	"time" 	"github.com/dustin/go-humanize"	"github.com/minio/madmin-go"	"github.com/minio/minio-go/v7"	miniogo "github.com/minio/minio-go/v7"	"github.com/minio/minio-go/v7/pkg/encrypt"	"github.com/minio/minio-go/v7/pkg/tags"	"github.com/minio/minio/internal/bucket/bandwidth"	"github.com/minio/minio/internal/bucket/replication"	"github.com/minio/minio/internal/config/storageclass"	"github.com/minio/minio/internal/crypto"	"github.com/minio/minio/internal/event"	"github.com/minio/minio/internal/hash"	xhttp "github.com/minio/minio/internal/http"	"github.com/minio/minio/internal/logger") const (	throttleDeadline = 1 * time.Hour	// ReplicationReset has reset id and timestamp of last reset operation	ReplicationReset = "replication-reset"	// ReplicationStatus has internal replication status - stringified representation of target's replication status for all replication	// activity initiated from this cluster	ReplicationStatus = "replication-status"	// ReplicationTimestamp - the last time replication was initiated on this cluster for this object version	ReplicationTimestamp = "replication-timestamp"	// ReplicaStatus - this header is present if a replica was received by this cluster for this object version	ReplicaStatus = "replica-status"	// ReplicaTimestamp - the last time a replica was received by this cluster for this object version	ReplicaTimestamp = "replica-timestamp"	// TaggingTimestamp - the last time a tag metadata modification happened on this cluster for this object version	TaggingTimestamp = "tagging-timestamp"	// ObjectLockRetentionTimestamp - the last time a object lock metadata modification happened on this cluster for this object version	ObjectLockRetentionTimestamp = "objectlock-retention-timestamp"	// ObjectLockLegalHoldTimestamp - the last time a legal hold metadata modification happened on this cluster for this object version	ObjectLockLegalHoldTimestamp = "objectlock-legalhold-timestamp"	// ReplicationWorkerMultiplier is suggested worker multiplier if traffic exceeds replication worker capacity	ReplicationWorkerMultiplier = 1.5) // gets replication config associated to a given bucket name.func getReplicationConfig(ctx context.Context, bucketName string) (rc *replication.Config, err error) {	if globalIsGateway {		objAPI := newObjectLayerFn()		if objAPI == nil {			return rc, errServerNotInitialized		} 		return rc, BucketReplicationConfigNotFound{Bucket: bucketName}	} 	rCfg, _, err := globalBucketMetadataSys.GetReplicationConfig(ctx, bucketName)	return rCfg, err} // validateReplicationDestination returns error if replication destination bucket missing or not configured// It also returns true if replication destination is same as this server.func validateReplicationDestination(ctx context.Context, bucket string, rCfg *replication.Config, checkRemote bool) (bool, APIError) {	var arns []string	if rCfg.RoleArn != "" {		arns = append(arns, rCfg.RoleArn)	} else {		for _, rule := range rCfg.Rules {			arns = append(arns, rule.Destination.String())		}	}	var sameTarget bool	for _, arnStr := range arns {		arn, err := madmin.ParseARN(arnStr)		if err != nil {			return sameTarget, errorCodes.ToAPIErrWithErr(ErrBucketRemoteArnInvalid, err)		}		if arn.Type != madmin.ReplicationService {			return sameTarget, toAPIError(ctx, BucketRemoteArnTypeInvalid{Bucket: bucket})		}		clnt := globalBucketTargetSys.GetRemoteTargetClient(ctx, arnStr)		if clnt == nil {			return sameTarget, toAPIError(ctx, BucketRemoteTargetNotFound{Bucket: bucket})		}		if checkRemote { // validate remote bucket			if found, err := clnt.BucketExists(ctx, arn.Bucket); !found {				return sameTarget, errorCodes.ToAPIErrWithErr(ErrRemoteDestinationNotFoundError, err)			}			if ret, err := globalBucketObjectLockSys.Get(bucket); err == nil {				if ret.LockEnabled {					lock, _, _, _, err := clnt.GetObjectLockConfig(ctx, arn.Bucket)					if err != nil || lock != "Enabled" {						return sameTarget, errorCodes.ToAPIErrWithErr(ErrReplicationDestinationMissingLock, err)					}				}			}		}		// validate replication ARN against target endpoint		c, ok := globalBucketTargetSys.arnRemotesMap[arnStr]		if ok {			if c.EndpointURL().String() == clnt.EndpointURL().String() {				selfTarget, _ := isLocalHost(clnt.EndpointURL().Hostname(), clnt.EndpointURL().Port(), globalMinioPort)				if !sameTarget {					sameTarget = selfTarget				}				continue			}		}	} 	if len(arns) == 0 {		return false, toAPIError(ctx, BucketRemoteTargetNotFound{Bucket: bucket})	}	return sameTarget, toAPIError(ctx, nil)} type mustReplicateOptions struct {	meta               map[string]string	status             replication.StatusType	opType             replication.Type	replicationRequest bool // incoming request is a replication request} func (o mustReplicateOptions) ReplicationStatus() (s replication.StatusType) {	if rs, ok := o.meta[xhttp.AmzBucketReplicationStatus]; ok {		return replication.StatusType(rs)	}	return s} func (o mustReplicateOptions) isExistingObjectReplication() bool {	return o.opType == replication.ExistingObjectReplicationType} func (o mustReplicateOptions) isMetadataReplication() bool {	return o.opType == replication.MetadataReplicationType} func getMustReplicateOptions(o ObjectInfo, op replication.Type, opts ObjectOptions) mustReplicateOptions {	if !op.Valid() {		op = replication.ObjectReplicationType		if o.metadataOnly {			op = replication.MetadataReplicationType		}	}	meta := cloneMSS(o.UserDefined)	if o.UserTags != "" {		meta[xhttp.AmzObjectTagging] = o.UserTags	} 	return mustReplicateOptions{		meta:               meta,		status:             o.ReplicationStatus,		opType:             op,		replicationRequest: opts.ReplicationRequest,	}} // mustReplicate returns 2 booleans - true if object meets replication criteria and true if replication is to be done in// a synchronous manner.func mustReplicate(ctx context.Context, bucket, object string, mopts mustReplicateOptions) (dsc ReplicateDecision) {	if globalIsGateway {		return	} 	// object layer not initialized we return with no decision.	if newObjectLayerFn() == nil {		return	} 	// Disable server-side replication on object prefixes which are excluded	// from versioning via the MinIO bucket versioning extension.	if globalBucketVersioningSys.PrefixSuspended(bucket, object) {		return	} 	replStatus := mopts.ReplicationStatus()	if replStatus == replication.Replica && !mopts.isMetadataReplication() {		return	} 	if mopts.replicationRequest { // incoming replication request on target cluster		return	}	cfg, err := getReplicationConfig(ctx, bucket)	if err != nil {		return	}	opts := replication.ObjectOpts{		Name:           object,		SSEC:           crypto.SSEC.IsEncrypted(mopts.meta),		Replica:        replStatus == replication.Replica,		ExistingObject: mopts.isExistingObjectReplication(),	}	tagStr, ok := mopts.meta[xhttp.AmzObjectTagging]	if ok {		opts.UserTags = tagStr	}	tgtArns := cfg.FilterTargetArns(opts)	for _, tgtArn := range tgtArns {		tgt := globalBucketTargetSys.GetRemoteTargetClient(ctx, tgtArn)		// the target online status should not be used here while deciding		// whether to replicate as the target could be temporarily down		opts.TargetArn = tgtArn		replicate := cfg.Replicate(opts)		var synchronous bool		if tgt != nil {			synchronous = tgt.replicateSync		}		dsc.Set(newReplicateTargetDecision(tgtArn, replicate, synchronous))	}	return dsc} // Standard headers that needs to be extracted from User metadata.var standardHeaders = []string{	xhttp.ContentType,	xhttp.CacheControl,	xhttp.ContentEncoding,	xhttp.ContentLanguage,	xhttp.ContentDisposition,	xhttp.AmzStorageClass,	xhttp.AmzObjectTagging,	xhttp.AmzBucketReplicationStatus,	xhttp.AmzObjectLockMode,	xhttp.AmzObjectLockRetainUntilDate,	xhttp.AmzObjectLockLegalHold,	xhttp.AmzTagCount,	xhttp.AmzServerSideEncryption,} // returns true if any of the objects being deleted qualifies for replication.func hasReplicationRules(ctx context.Context, bucket string, objects []ObjectToDelete) bool {	c, err := getReplicationConfig(ctx, bucket)	if err != nil || c == nil {		return false	}	for _, obj := range objects {		if c.HasActiveRules(obj.ObjectName, true) {			return true		}	}	return false} // isStandardHeader returns true if header is a supported header and not a custom headerfunc isStandardHeader(matchHeaderKey string) bool {	return equals(matchHeaderKey, standardHeaders...)} // returns whether object version is a deletemarker and if object qualifies for replicationfunc checkReplicateDelete(ctx context.Context, bucket string, dobj ObjectToDelete, oi ObjectInfo, delOpts ObjectOptions, gerr error) (dsc ReplicateDecision) {	rcfg, err := getReplicationConfig(ctx, bucket)	if err != nil || rcfg == nil {		return	}	// If incoming request is a replication request, it does not need to be re-replicated.	if delOpts.ReplicationRequest {		return	}	// Skip replication if this object's prefix is excluded from being	// versioned.	if !delOpts.Versioned {		return	}	opts := replication.ObjectOpts{		Name:         dobj.ObjectName,		SSEC:         crypto.SSEC.IsEncrypted(oi.UserDefined),		UserTags:     oi.UserTags,		DeleteMarker: oi.DeleteMarker,		VersionID:    dobj.VersionID,		OpType:       replication.DeleteReplicationType,	}	tgtArns := rcfg.FilterTargetArns(opts)	if len(tgtArns) > 0 {		dsc.targetsMap = make(map[string]replicateTargetDecision, len(tgtArns))		var sync, replicate bool		for _, tgtArn := range tgtArns {			opts.TargetArn = tgtArn			replicate = rcfg.Replicate(opts)			// when incoming delete is removal of a delete marker( a.k.a versioned delete),			// GetObjectInfo returns extra information even though it returns errFileNotFound			if gerr != nil {				validReplStatus := false				switch oi.TargetReplicationStatus(tgtArn) {				case replication.Pending, replication.Completed, replication.Failed:					validReplStatus = true				}				if oi.DeleteMarker && (validReplStatus || replicate) {					dsc.Set(newReplicateTargetDecision(tgtArn, replicate, sync))					continue				} else {					// can be the case that other cluster is down and duplicate `mc rm --vid`					// is issued - this still needs to be replicated back to the other target					replicate = oi.VersionPurgeStatus == Pending || oi.VersionPurgeStatus == Failed					dsc.Set(newReplicateTargetDecision(tgtArn, replicate, sync))					continue				}			}			tgt := globalBucketTargetSys.GetRemoteTargetClient(ctx, tgtArn)			// the target online status should not be used here while deciding			// whether to replicate deletes as the target could be temporarily down			tgtDsc := newReplicateTargetDecision(tgtArn, false, false)			if tgt != nil {				tgtDsc = newReplicateTargetDecision(tgtArn, replicate, tgt.replicateSync)			}			dsc.Set(tgtDsc)		}	}	return dsc} // replicate deletes to the designated replication target if replication configuration// has delete marker replication or delete replication (MinIO extension to allow deletes where version id// is specified) enabled.// Similar to bucket replication for PUT operation, soft delete (a.k.a setting delete marker) and// permanent deletes (by specifying a version ID in the delete operation) have three states "Pending", "Complete"// and "Failed" to mark the status of the replication of "DELETE" operation. All failed operations can// then be retried by healing. In the case of permanent deletes, until the replication is completed on the// target cluster, the object version is marked deleted on the source and hidden from listing. It is permanently// deleted from the source when the VersionPurgeStatus changes to "Complete", i.e after replication succeeds// on target.func replicateDelete(ctx context.Context, dobj DeletedObjectReplicationInfo, objectAPI ObjectLayer) {	var replicationStatus replication.StatusType	bucket := dobj.Bucket	versionID := dobj.DeleteMarkerVersionID	if versionID == "" {		versionID = dobj.VersionID	} 	defer func() {		replStatus := string(replicationStatus)		auditLogInternal(context.Background(), bucket, dobj.ObjectName, AuditLogOptions{			Event:     dobj.EventType,			APIName:   ReplicateDeleteAPI,			VersionID: versionID,			Status:    replStatus,		})	}() 	rcfg, err := getReplicationConfig(ctx, bucket)	if err != nil || rcfg == nil {		logger.LogIf(ctx, err)		sendEvent(eventArgs{			BucketName: bucket,			Object: ObjectInfo{				Bucket:       bucket,				Name:         dobj.ObjectName,				VersionID:    versionID,				DeleteMarker: dobj.DeleteMarker,			},			Host:      "Internal: [Replication]",			EventName: event.ObjectReplicationNotTracked,		})		return	}	dsc, err := parseReplicateDecision(dobj.ReplicationState.ReplicateDecisionStr)	if err != nil {		logger.LogIf(ctx, err)		sendEvent(eventArgs{			BucketName: bucket,			Object: ObjectInfo{				Bucket:       bucket,				Name:         dobj.ObjectName,				VersionID:    versionID,				DeleteMarker: dobj.DeleteMarker,			},			Host:      "Internal: [Replication]",			EventName: event.ObjectReplicationNotTracked,		})		return	} 	// Lock the object name before starting replication operation.	// Use separate lock that doesn't collide with regular objects.	lk := objectAPI.NewNSLock(bucket, "/[replicate]/"+dobj.ObjectName)	lkctx, err := lk.GetLock(ctx, globalOperationTimeout)	if err != nil {		logger.LogIf(ctx, fmt.Errorf("failed to get lock for object: %s bucket:%s arn:%s", dobj.ObjectName, bucket, rcfg.RoleArn))		sendEvent(eventArgs{			BucketName: bucket,			Object: ObjectInfo{				Bucket:       bucket,				Name:         dobj.ObjectName,				VersionID:    versionID,				DeleteMarker: dobj.DeleteMarker,			},			Host:      "Internal: [Replication]",			EventName: event.ObjectReplicationNotTracked,		})		return	}	ctx = lkctx.Context()	defer lk.Unlock(lkctx.Cancel) 	var wg sync.WaitGroup	var rinfos replicatedInfos	rinfos.Targets = make([]replicatedTargetInfo, len(dsc.targetsMap))	idx := -1	for tgtArn := range dsc.targetsMap {		idx++		tgt := globalBucketTargetSys.GetRemoteTargetClient(ctx, tgtArn)		if tgt == nil {			logger.LogIf(ctx, fmt.Errorf("failed to get target for bucket:%s arn:%s", bucket, tgtArn))			sendEvent(eventArgs{				BucketName: bucket,				Object: ObjectInfo{					Bucket:       bucket,					Name:         dobj.ObjectName,					VersionID:    versionID,					DeleteMarker: dobj.DeleteMarker,				},				Host:      "Internal: [Replication]",				EventName: event.ObjectReplicationNotTracked,			})			continue		}		if tgt := dsc.targetsMap[tgtArn]; !tgt.Replicate {			continue		}		// if dobj.TargetArn is not empty string, this is a case of specific target being re-synced.		if dobj.TargetArn != "" && dobj.TargetArn != tgt.ARN {			continue		}		wg.Add(1)		go func(index int, tgt *TargetClient) {			defer wg.Done()			rinfo := replicateDeleteToTarget(ctx, dobj, tgt)			rinfos.Targets[index] = rinfo		}(idx, tgt)	}	wg.Wait() 	replicationStatus = rinfos.ReplicationStatus()	prevStatus := dobj.DeleteMarkerReplicationStatus() 	if dobj.VersionID != "" {		prevStatus = replication.StatusType(dobj.VersionPurgeStatus())		replicationStatus = replication.StatusType(rinfos.VersionPurgeStatus())	} 	// to decrement pending count later.	for _, rinfo := range rinfos.Targets {		if rinfo.ReplicationStatus != rinfo.PrevReplicationStatus {			globalReplicationStats.Update(dobj.Bucket, rinfo.Arn, 0, 0, replicationStatus,				prevStatus, replication.DeleteReplicationType)		}	} 	eventName := event.ObjectReplicationComplete	if replicationStatus == replication.Failed {		eventName = event.ObjectReplicationFailed	}	drs := getReplicationState(rinfos, dobj.ReplicationState, dobj.VersionID)	if replicationStatus != prevStatus {		drs.ReplicationTimeStamp = UTCNow()	} 	dobjInfo, err := objectAPI.DeleteObject(ctx, bucket, dobj.ObjectName, ObjectOptions{		VersionID:         versionID,		MTime:             dobj.DeleteMarkerMTime.Time,		DeleteReplication: drs,		Versioned:         globalBucketVersioningSys.PrefixEnabled(bucket, dobj.ObjectName),		VersionSuspended:  globalBucketVersioningSys.PrefixSuspended(bucket, dobj.ObjectName),	})	if err != nil && !isErrVersionNotFound(err) { // VersionNotFound would be reported by pool that object version is missing on.		logger.LogIf(ctx, fmt.Errorf("Unable to update replication metadata for %s/%s(%s): %s", bucket, dobj.ObjectName, versionID, err))		sendEvent(eventArgs{			BucketName: bucket,			Object: ObjectInfo{				Bucket:       bucket,				Name:         dobj.ObjectName,				VersionID:    versionID,

Lines 1–500 of 2388. Load more