oyd
/
minio
6
1
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Revert "Add delete marker replication support (#10396)"

Browse Source 
This reverts commit 50c10a5087.

PR is moved to origin/dev branch
master
Harshavardhana 4 years ago
parent be83dfc52a
commit 267d7bf0a9
25 changed files with 227 additions and 950 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 13
      cmd/api-datatypes.go
  2. 25
      cmd/bucket-handlers.go
  3. 157
      cmd/bucket-replication.go
  4. 35
      cmd/data-crawler.go
  5. 5
      cmd/erasure-metadata.go
  6. 130
      cmd/erasure-object.go
  7. 6
      cmd/http/headers.go
  8. 2
      cmd/metacache-entries.go
  9. 3
      cmd/object-api-datatypes.go
  10. 21
      cmd/object-api-interface.go
  11. 61
      cmd/object-api-options.go
  12. 40
      cmd/object-handlers.go
  13. 33
      cmd/storage-datatypes.go
  14. 128
      cmd/storage-datatypes_gen.go
  15. 74
      cmd/web-handlers.go
  16. 4
      cmd/xl-storage-format-utils.go
  17. 74
      cmd/xl-storage-format-v2.go
  18. 230
      cmd/xl-storage-format-v2_gen.go
  19. 6
      cmd/xl-storage.go
  20. BIN
      docs/bucket/replication/DELETE_bucket-replication.png
  21. 29
      docs/bucket/replication/README.md
  22. 3
      go.mod
  23. 32
      go.sum
  24. 15
      pkg/bucket/replication/replication.go
  25. 51
      pkg/bucket/replication/rule.go

13
cmd/api-datatypes.go
Unescape View File

@ -18,7 +18,6 @@ package cmd
import (
"encoding/xml"
"time"
)
// DeletedObject objects deleted
@ -27,24 +26,12 @@ type DeletedObject struct {
DeleteMarkerVersionID string `xml:"DeleteMarkerVersionId,omitempty"`
ObjectName string `xml:"Key,omitempty"`
VersionID string `xml:"VersionId,omitempty"`
// Replication status of DeleteMarker
DeleteMarkerReplicationStatus string
// MTime of DeleteMarker on source that needs to be propagated to replica
DeleteMarkerMTime time.Time
// Status of versioned delete (of object or DeleteMarker)
VersionPurgeStatus VersionPurgeStatusType
}
// ObjectToDelete carries key name for the object to delete.
type ObjectToDelete struct {
ObjectName string `xml:"Key"`
VersionID string `xml:"VersionId"`
// Replication status of DeleteMarker
DeleteMarkerReplicationStatus string
// Status of versioned delete (of object or DeleteMarker)
VersionPurgeStatus VersionPurgeStatusType
// Version ID of delete marker
DeleteMarkerVersionID string
}
// createBucketConfiguration container for bucket configuration request from client.

25
cmd/bucket-handlers.go
Unescape View File

@ -404,7 +404,6 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
if api.CacheAPI() != nil {
getObjectInfoFn = api.CacheAPI().GetObjectInfo
}
replicateDeletes := hasReplicationRules(ctx, bucket, deleteObjects.Objects)
dErrs := make([]DeleteError, len(deleteObjects.Objects))
for index, object := range deleteObjects.Objects {
@ -440,18 +439,6 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
// Avoid duplicate objects, we use map to filter them out.
if _, ok := objectsToDelete[object]; !ok {
if replicateDeletes {
if delMarker, replicate := checkReplicateDelete(ctx, getObjectInfoFn, bucket, ObjectToDelete{ObjectName: object.ObjectName, VersionID: object.VersionID}); replicate {
if object.VersionID != "" {
object.VersionPurgeStatus = Pending
if delMarker {
object.DeleteMarkerVersionID = object.VersionID
}
} else {
object.DeleteMarkerReplicationStatus = string(replication.Pending)
}
}
}
objectsToDelete[object] = index
}
}
@ -480,8 +467,6 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
}]
apiErr := toAPIError(ctx, errs[i])
if apiErr.Code == "" || apiErr.Code == "NoSuchKey" || apiErr.Code == "InvalidArgument" {
dObjects[i].DeleteMarkerReplicationStatus = deleteList[i].DeleteMarkerReplicationStatus
dObjects[i].VersionPurgeStatus = deleteList[i].VersionPurgeStatus
deletedObjects[dindex] = dObjects[i]
continue
}
@ -506,14 +491,7 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
// Write success response.
writeSuccessResponseXML(w, encodedSuccessResponse)
for _, dobj := range deletedObjects {
if dobj.DeleteMarkerReplicationStatus == string(replication.Pending) || dobj.VersionPurgeStatus == Pending {
globalReplicationState.queueReplicaDeleteTask(DeletedObjectVersionInfo{
DeletedObject: dobj,
Bucket: bucket,
})
}
}
// Notify deleted event for objects.
for _, dobj := range deletedObjects {
eventName := event.ObjectRemovedDelete
@ -1320,6 +1298,7 @@ func (api objectAPIHandlers) PutBucketReplicationConfigHandler(w http.ResponseWr
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
return
}
if err = globalBucketMetadataSys.Update(bucket, bucketReplicationConfig, configData); err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
return

157
cmd/bucket-replication.go
Unescape View File

@ -83,7 +83,7 @@ func mustReplicateWeb(ctx context.Context, r *http.Request, bucket, object strin
if permErr != ErrNone {
return false
}
return mustReplicater(ctx, bucket, object, meta, replStatus)
return mustReplicater(ctx, r, bucket, object, meta, replStatus)
}
// mustReplicate returns true if object meets replication criteria.
@ -91,11 +91,11 @@ func mustReplicate(ctx context.Context, r *http.Request, bucket, object string,
if s3Err := isPutActionAllowed(ctx, getRequestAuthType(r), bucket, "", r, iampolicy.GetReplicationConfigurationAction); s3Err != ErrNone {
return false
}
return mustReplicater(ctx, bucket, object, meta, replStatus)
return mustReplicater(ctx, r, bucket, object, meta, replStatus)
}
// mustReplicater returns true if object meets replication criteria.
func mustReplicater(ctx context.Context, bucket, object string, meta map[string]string, replStatus string) bool {
func mustReplicater(ctx context.Context, r *http.Request, bucket, object string, meta map[string]string, replStatus string) bool {
if globalIsGateway {
return false
}
@ -120,127 +120,6 @@ func mustReplicater(ctx context.Context, bucket, object string, meta map[string]
return cfg.Replicate(opts)
}
// 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
}
// returns whether object version is a deletemarker and if object qualifies for replication
func checkReplicateDelete(ctx context.Context, getObjectInfoFn GetObjectInfoFn, bucket string, dobj ObjectToDelete) (dm, replicate bool) {
rcfg, err := getReplicationConfig(ctx, bucket)
if err != nil || rcfg == nil {
return false, false
}
oi, err := getObjectInfoFn(ctx, bucket, dobj.ObjectName, ObjectOptions{VersionID: dobj.VersionID})
// when incoming delete is removal of a delete marker( a.k.a versioned delete),
// GetObjectInfo returns extra information even though it returns errFileNotFound
if err != nil {
validReplStatus := false
switch oi.ReplicationStatus {
case replication.Pending, replication.Complete, replication.Failed:
validReplStatus = true
}
if oi.DeleteMarker && validReplStatus {
return oi.DeleteMarker, true
}
return oi.DeleteMarker, false
}
opts := replication.ObjectOpts{
Name: dobj.ObjectName,
SSEC: crypto.SSEC.IsEncrypted(oi.UserDefined),
UserTags: oi.UserTags,
DeleteMarker: true,
VersionID: dobj.VersionID,
}
return oi.DeleteMarker, rcfg.Replicate(opts)
}
// 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 DeletedObjectVersionInfo, objectAPI ObjectLayer) {
bucket := dobj.Bucket
rcfg, err := getReplicationConfig(ctx, bucket)
if err != nil || rcfg == nil {
return
}
tgt := globalBucketTargetSys.GetRemoteTargetClient(ctx, rcfg.RoleArn)
if tgt == nil {
return
}
versionID := dobj.DeleteMarkerVersionID
if versionID == "" {
versionID = dobj.VersionID
}
rmErr := tgt.RemoveObject(ctx, rcfg.GetDestination().Bucket, dobj.ObjectName, miniogo.RemoveObjectOptions{
VersionID: versionID,
Internal: miniogo.AdvancedRemoveOptions{
ReplicationDeleteMarker: dobj.DeleteMarkerVersionID != "",
ReplicationMTime: dobj.DeleteMarkerMTime,
ReplicationStatus: miniogo.ReplicationStatusReplica,
},
})
replicationStatus := dobj.DeleteMarkerReplicationStatus
versionPurgeStatus := dobj.VersionPurgeStatus
if rmErr != nil {
if dobj.VersionID == "" {
replicationStatus = string(replication.Failed)
} else {
versionPurgeStatus = Failed
}
} else {
if dobj.VersionID == "" {
replicationStatus = string(replication.Complete)
} else {
versionPurgeStatus = Complete
}
}
if replicationStatus == string(replication.Failed) || versionPurgeStatus == Failed {
objInfo := ObjectInfo{
Name: dobj.ObjectName,
DeleteMarker: dobj.DeleteMarker,
VersionID: versionID,
ReplicationStatus: replication.StatusType(dobj.DeleteMarkerReplicationStatus),
}
eventArg := &eventArgs{
BucketName: bucket,
Object: objInfo,
Host: "Internal: [Replication]",
EventName: event.ObjectReplicationFailed,
}
sendEvent(*eventArg)
}
// Update metadata on the delete marker or purge permanent delete if replication success.
if _, err = objectAPI.DeleteObject(ctx, bucket, dobj.ObjectName, ObjectOptions{
VersionID: versionID,
DeleteMarker: dobj.DeleteMarker,
DeleteMarkerReplicationStatus: replicationStatus,
Versioned: globalBucketVersioningSys.Enabled(bucket),
VersionPurgeStatus: versionPurgeStatus,
VersionSuspended: globalBucketVersioningSys.Suspended(bucket),
}); err != nil {
logger.LogIf(ctx, fmt.Errorf("Unable to update replication metadata for %s/%s %s: %w", bucket, dobj.ObjectName, dobj.VersionID, err))
}
}
func putReplicationOpts(ctx context.Context, dest replication.Destination, objInfo ObjectInfo) (putOpts miniogo.PutObjectOptions) {
meta := make(map[string]string)
for k, v := range objInfo.UserDefined {
@ -424,37 +303,18 @@ func filterReplicationStatusMetadata(metadata map[string]string) map[string]stri
return dst
}
// DeletedObjectVersionInfo has info on deleted object
type DeletedObjectVersionInfo struct {
DeletedObject
Bucket string
}
type replicationState struct {
// add future metrics here
replicaCh chan ObjectInfo
replicaDeleteCh chan DeletedObjectVersionInfo
replicaCh chan ObjectInfo
}
func (r *replicationState) queueReplicaTask(oi ObjectInfo) {
if r == nil {
return
}
select {
case r.replicaCh <- oi:
default:
}
}
func (r *replicationState) queueReplicaDeleteTask(doi DeletedObjectVersionInfo) {
if r == nil {
return
}
select {
case r.replicaDeleteCh <- doi:
default:
}
}
var (
globalReplicationState *replicationState
// TODO: currently keeping it conservative
@ -471,13 +331,11 @@ func newReplicationState() *replicationState {
globalReplicationConcurrent = 1
}
rs := &replicationState{
replicaCh: make(chan ObjectInfo, 10000),
replicaDeleteCh: make(chan DeletedObjectVersionInfo, 10000),
replicaCh: make(chan ObjectInfo, 10000),
}
go func() {
<-GlobalContext.Done()
close(rs.replicaCh)
close(rs.replicaDeleteCh)
}()
return rs
}
@ -495,11 +353,6 @@ func (r *replicationState) addWorker(ctx context.Context, objectAPI ObjectLayer)
return
}
replicateObject(ctx, oi, objectAPI)
case doi, ok := <-r.replicaDeleteCh:
if !ok {
return
}
replicateDelete(ctx, doi, objectAPI)
}
}
}()

35
cmd/data-crawler.go
Unescape View File

@ -792,43 +792,8 @@ func sleepDuration(d time.Duration, x float64) {
// healReplication will heal a scanned item that has failed replication.
func (i *crawlItem) healReplication(ctx context.Context, o ObjectLayer, meta actionMeta) {
if meta.oi.DeleteMarker || !meta.oi.VersionPurgeStatus.Empty() {
//heal delete marker replication failure or versioned delete replication failure
if meta.oi.ReplicationStatus == replication.Pending ||
meta.oi.ReplicationStatus == replication.Failed ||
meta.oi.VersionPurgeStatus == Failed || meta.oi.VersionPurgeStatus == Pending {
i.healReplicationDeletes(ctx, o, meta)
return
}
}
if meta.oi.ReplicationStatus == replication.Pending ||
meta.oi.ReplicationStatus == replication.Failed {
globalReplicationState.queueReplicaTask(meta.oi)
}
}
// healReplicationDeletes will heal a scanned deleted item that failed to replicate deletes.
func (i *crawlItem) healReplicationDeletes(ctx context.Context, o ObjectLayer, meta actionMeta) {
// handle soft delete and permanent delete failures here.
if meta.oi.DeleteMarker || !meta.oi.VersionPurgeStatus.Empty() {
versionID := ""
dmVersionID := ""
if meta.oi.VersionPurgeStatus.Empty() {
dmVersionID = meta.oi.VersionID
} else {
versionID = meta.oi.VersionID
}
globalReplicationState.queueReplicaDeleteTask(DeletedObjectVersionInfo{
DeletedObject: DeletedObject{
ObjectName: meta.oi.Name,
DeleteMarkerVersionID: dmVersionID,
VersionID: versionID,
DeleteMarkerReplicationStatus: string(meta.oi.ReplicationStatus),
DeleteMarkerMTime: meta.oi.ModTime,
DeleteMarker: meta.oi.DeleteMarker,
VersionPurgeStatus: meta.oi.VersionPurgeStatus,
},
Bucket: meta.oi.Bucket,
})
}
}

5
cmd/erasure-metadata.go
Unescape View File

@ -139,9 +139,7 @@ func (fi FileInfo) ToObjectInfo(bucket, object string) ObjectInfo {
// Add replication status to the object info
objInfo.ReplicationStatus = replication.StatusType(fi.Metadata[xhttp.AmzBucketReplicationStatus])
if fi.Deleted {
objInfo.ReplicationStatus = replication.StatusType(fi.DeleteMarkerReplicationStatus)
}
// etag/md5Sum has already been extracted. We need to
// remove to avoid it from appearing as part of
// response headers. e.g, X-Minio-* or X-Amz-*.
@ -157,7 +155,6 @@ func (fi FileInfo) ToObjectInfo(bucket, object string) ObjectInfo {
} else {
objInfo.StorageClass = globalMinioDefaultStorageClass
}
objInfo.VersionPurgeStatus = fi.VersionPurgeStatus
// Success.
return objInfo
}

130
cmd/erasure-object.go
Unescape View File

@ -29,7 +29,6 @@ import (
"github.com/minio/minio-go/v7/pkg/tags"
xhttp "github.com/minio/minio/cmd/http"
"github.com/minio/minio/cmd/logger"
"github.com/minio/minio/pkg/bucket/replication"
"github.com/minio/minio/pkg/mimedb"
"github.com/minio/minio/pkg/sync/errgroup"
)
@ -406,7 +405,7 @@ func (er erasureObjects) getObjectInfo(ctx context.Context, bucket, object strin
if fi.Deleted {
objInfo = fi.ToObjectInfo(bucket, object)
if opts.VersionID == "" || opts.DeleteMarker {
if opts.VersionID == "" {
return objInfo, toObjectErr(errFileNotFound, bucket, object)
}
// Make sure to return object info to provide extra information.
@ -739,6 +738,7 @@ func (er erasureObjects) deleteObjectVersion(ctx context.Context, bucket, object
return disks[index].DeleteVersion(ctx, bucket, object, fi)
}, index)
}
// return errors if any during deletion
return reduceWriteQuorumErrs(ctx, g.Wait(), objectOpIgnoredErrs, writeQuorum)
}
@ -810,35 +810,26 @@ func (er erasureObjects) DeleteObjects(ctx context.Context, bucket string, objec
versions := make([]FileInfo, len(objects))
for i := range objects {
modTime := opts.MTime
if opts.MTime.IsZero() {
modTime = UTCNow()
}
uuid := opts.VersionID
if uuid == "" {
uuid = mustGetUUID()
}
if objects[i].VersionID == "" {
if (opts.Versioned || opts.VersionSuspended) && !HasSuffix(objects[i].ObjectName, SlashSeparator) {
versions[i] = FileInfo{
Name: objects[i].ObjectName,
ModTime: modTime,
Deleted: true, // delete marker
DeleteMarkerReplicationStatus: objects[i].DeleteMarkerReplicationStatus,
VersionPurgeStatus: objects[i].VersionPurgeStatus,
if opts.Versioned || opts.VersionSuspended {
fi := FileInfo{
Name: objects[i].ObjectName,
ModTime: UTCNow(),
Deleted: true, // delete marker
}
if opts.Versioned {
versions[i].VersionID = uuid
fi.VersionID = mustGetUUID()
}
// versioning suspended means we add `null`
// version as delete marker
versions[i] = fi
continue
}
}
versions[i] = FileInfo{
Name: objects[i].ObjectName,
VersionID: objects[i].VersionID,
DeleteMarkerReplicationStatus: objects[i].DeleteMarkerReplicationStatus,
VersionPurgeStatus: objects[i].VersionPurgeStatus,
Name: objects[i].ObjectName,
VersionID: objects[i].VersionID,
}
}
@ -885,19 +876,14 @@ func (er erasureObjects) DeleteObjects(ctx context.Context, bucket string, objec
ObjectPathUpdated(pathJoin(bucket, objects[objIndex].ObjectName))
if versions[objIndex].Deleted {
dobjects[objIndex] = DeletedObject{
DeleteMarker: versions[objIndex].Deleted,
DeleteMarkerVersionID: versions[objIndex].VersionID,
DeleteMarkerMTime: versions[objIndex].ModTime,
DeleteMarkerReplicationStatus: versions[objIndex].DeleteMarkerReplicationStatus,
ObjectName: versions[objIndex].Name,
VersionPurgeStatus: versions[objIndex].VersionPurgeStatus,
DeleteMarker: versions[objIndex].Deleted,
DeleteMarkerVersionID: versions[objIndex].VersionID,
ObjectName: decodeDirObject(versions[objIndex].Name),
}
} else {
dobjects[objIndex] = DeletedObject{
ObjectName: versions[objIndex].Name,
VersionID: versions[objIndex].VersionID,
VersionPurgeStatus: versions[objIndex].VersionPurgeStatus,
DeleteMarkerReplicationStatus: versions[objIndex].DeleteMarkerReplicationStatus,
ObjectName: decodeDirObject(versions[objIndex].Name),
VersionID: versions[objIndex].VersionID,
}
}
}
@ -927,20 +913,16 @@ func (er erasureObjects) DeleteObjects(ctx context.Context, bucket string, objec
// any error as it is not necessary for the handler to reply back a
// response to the client request.
func (er erasureObjects) DeleteObject(ctx context.Context, bucket, object string, opts ObjectOptions) (objInfo ObjectInfo, err error) {
versionFound := true
defer ObjectPathUpdated(path.Join(bucket, object))
goi, gerr := er.GetObjectInfo(ctx, bucket, object, opts)
if gerr != nil && goi.Name == "" {
switch gerr.(type) {
case InsufficientReadQuorum:
return objInfo, InsufficientWriteQuorum{}
}
// For delete marker replication, versionID being replicated will not exist on disk
if opts.DeleteMarker {
versionFound = false
} else {
return objInfo, gerr
}
return objInfo, gerr
}
// Acquire a write lock before deleting the object.
lk := er.NewNSLock(bucket, object)
if err = lk.GetLock(ctx, globalDeleteOperationTimeout); err != nil {
@ -950,58 +932,23 @@ func (er erasureObjects) DeleteObject(ctx context.Context, bucket, object string
storageDisks := er.getDisks()
writeQuorum := len(storageDisks)/2 + 1
var markDelete bool
// Determine whether to mark object deleted for replication
if goi.VersionID != "" {
markDelete = true
}
// Default deleteMarker to true if object is under versioning
deleteMarker := true
if gerr == nil {
deleteMarker = goi.VersionID != ""
}
if opts.VersionID != "" {
// case where replica version needs to be deleted on target cluster
if versionFound && opts.DeleteMarkerReplicationStatus == replication.Replica.String() {
markDelete = false
}
if opts.VersionPurgeStatus.Empty() && opts.DeleteMarkerReplicationStatus == "" {
markDelete = false
}
if opts.DeleteMarker && opts.VersionPurgeStatus == Complete {
markDelete = false
}
// determine if the version represents an object delete
// deleteMarker = true
if versionFound && !goi.DeleteMarker { // implies a versioned delete of object
deleteMarker = false
}
}
modTime := opts.MTime
if opts.MTime.IsZero() {
modTime = UTCNow()
}
if markDelete {
if (opts.Versioned || opts.VersionSuspended) && !HasSuffix(object, SlashSeparator) {
if opts.VersionID == "" {
if opts.Versioned || opts.VersionSuspended {
fi := FileInfo{
Name: object,
Deleted: deleteMarker,
MarkDeleted: markDelete,
ModTime: modTime,
DeleteMarkerReplicationStatus: opts.DeleteMarkerReplicationStatus,
VersionPurgeStatus: opts.VersionPurgeStatus,
Name: object,
Deleted: true,
ModTime: UTCNow(),
}
if opts.Versioned {
fi.VersionID = mustGetUUID()
if opts.VersionID != "" {
fi.VersionID = opts.VersionID
}
}
// versioning suspended means we add `null`
// version as delete marker
// Add delete marker, since we don't have any version specified explicitly.
// Or if a particular version id needs to be replicated.
if err = er.deleteObjectVersion(ctx, bucket, object, writeQuorum, fi); err != nil {
return objInfo, toObjectErr(err, bucket, object)
}
@ -1011,13 +958,8 @@ func (er erasureObjects) DeleteObject(ctx context.Context, bucket, object string
// Delete the object version on all disks.
if err = er.deleteObjectVersion(ctx, bucket, object, writeQuorum, FileInfo{
Name: object,
VersionID: opts.VersionID,
MarkDeleted: markDelete,
Deleted: deleteMarker,
ModTime: modTime,
DeleteMarkerReplicationStatus: opts.DeleteMarkerReplicationStatus,
VersionPurgeStatus: opts.VersionPurgeStatus,
Name: object,
VersionID: opts.VersionID,
}); err != nil {
return objInfo, toObjectErr(err, bucket, object)
}
@ -1029,13 +971,7 @@ func (er erasureObjects) DeleteObject(ctx context.Context, bucket, object string
}
}
return ObjectInfo{
Bucket: bucket,
Name: object,
VersionID: opts.VersionID,
VersionPurgeStatus: opts.VersionPurgeStatus,
ReplicationStatus: replication.StatusType(opts.DeleteMarkerReplicationStatus),
}, nil
return ObjectInfo{Bucket: bucket, Name: decodeDirObject(object), VersionID: opts.VersionID}, nil
}
// Send the successful but partial upload/delete, however ignore

6
cmd/http/headers.go
Unescape View File

@ -132,12 +132,6 @@ const (
// Reports number of drives currently healing
MinIOHealingDrives = "x-minio-healing-drives"
// Header indicates if the delete marker should be preserved by client
MinIOSourceDeleteMarker = "x-minio-source-deletemarker"
// Header indicates if the delete marker version needs to be purged.
MinIOSourceDeleteMarkerDelete = "x-minio-source-deletemarker-delete"
)
// Common http query params S3 API

2
cmd/metacache-entries.go
Unescape View File

@ -151,7 +151,7 @@ func (e *metaCacheEntry) fileInfoVersions(bucket string) (FileInfoVersions, erro
},
}, nil
}
return getFileInfoVersions(e.metadata, bucket, e.name, false)
return getFileInfoVersions(e.metadata, bucket, e.name)
}
// metaCacheEntries is a slice of metacache entries.

3
cmd/object-api-datatypes.go
Unescape View File

@ -207,8 +207,7 @@ type ObjectInfo struct {
Legacy bool // indicates object on disk is in legacy data format
// backendType indicates which backend filled this structure
backendType BackendType
VersionPurgeStatus VersionPurgeStatusType
backendType BackendType
}
// MultipartInfo captures metadata information about the uploadId

21
cmd/object-api-interface.go
Unescape View File

@ -36,18 +36,15 @@ type GetObjectInfoFn func(ctx context.Context, bucket, object string, opts Objec
// ObjectOptions represents object options for ObjectLayer object operations
type ObjectOptions struct {
ServerSideEncryption encrypt.ServerSide
VersionSuspended bool // indicates if the bucket was previously versioned but is currently suspended.
Versioned bool // indicates if the bucket is versioned
WalkVersions bool // indicates if the we are interested in walking versions
VersionID string // Specifies the versionID which needs to be overwritten or read
MTime time.Time // Is only set in POST/PUT operations
DeleteMarker bool // Is only set in DELETE operations for delete marker replication
UserDefined map[string]string // only set in case of POST/PUT operations
PartNumber int // only useful in case of GetObject/HeadObject
CheckPrecondFn CheckPreconditionFn // only set during GetObject/HeadObject/CopyObjectPart preconditional valuation
DeleteMarkerReplicationStatus string // Is only set in DELETE operations
VersionPurgeStatus VersionPurgeStatusType // Is only set in DELETE operations for delete marker version to be permanently deleted.
ServerSideEncryption encrypt.ServerSide
VersionSuspended bool // indicates if the bucket was previously versioned but is currently suspended.
Versioned bool // indicates if the bucket is versioned
WalkVersions bool // indicates if the we are interested in walking versions
VersionID string // Specifies the versionID which needs to be overwritten or read
MTime time.Time // Is only set in POST/PUT operations
UserDefined map[string]string // only set in case of POST/PUT operations
PartNumber int // only useful in case of GetObject/HeadObject
CheckPrecondFn CheckPreconditionFn // only set during GetObject/HeadObject/CopyObjectPart preconditional valuation
}
// BucketOptions represents bucket options for ObjectLayer bucket operations

61
cmd/object-api-options.go
Unescape View File

@ -123,20 +123,6 @@ func getOpts(ctx context.Context, r *http.Request, bucket, object string) (Objec
}
opts.PartNumber = partNumber
opts.VersionID = vid
delMarker := strings.TrimSpace(r.Header.Get(xhttp.MinIOSourceDeleteMarker))
if delMarker != "" {
if delMarker != "true" && delMarker != "false" {
logger.LogIf(ctx, err)
return opts, InvalidArgument{
Bucket: bucket,
Object: object,
Err: fmt.Errorf("Unable to parse %s, failed with %w", xhttp.MinIOSourceDeleteMarker, fmt.Errorf("DeleteMarker should be true or false")),
}
}
if delMarker == "true" {
opts.DeleteMarker = true
}
}
return opts, nil
}
@ -148,49 +134,6 @@ func delOpts(ctx context.Context, r *http.Request, bucket, object string) (opts
}
opts.Versioned = versioned
opts.VersionSuspended = globalBucketVersioningSys.Suspended(bucket)
delMarker := strings.TrimSpace(r.Header.Get(xhttp.MinIOSourceDeleteMarker))
if delMarker != "" {
if delMarker != "true" && delMarker != "false" {
logger.LogIf(ctx, err)
return opts, InvalidArgument{
Bucket: bucket,
Object: object,
Err: fmt.Errorf("Unable to parse %s, failed with %w", xhttp.MinIOSourceDeleteMarker, fmt.Errorf("DeleteMarker should be true or false")),
}
}
if delMarker == "true" {
opts.DeleteMarker = true
}
}
purgeVersion := strings.TrimSpace(r.Header.Get(xhttp.MinIOSourceDeleteMarkerDelete))
if purgeVersion != "" {
if purgeVersion != "true" && purgeVersion != "false" {
logger.LogIf(ctx, err)
return opts, InvalidArgument{
Bucket: bucket,
Object: object,
Err: fmt.Errorf("Unable to parse %s, failed with %w", xhttp.MinIOSourceDeleteMarkerDelete, fmt.Errorf("DeleteMarkerPurge should be true or false")),
}
}
if purgeVersion == "true" {
opts.VersionPurgeStatus = Complete
}
}
mtime := strings.TrimSpace(r.Header.Get(xhttp.MinIOSourceMTime))
if mtime != "" {
opts.MTime, err = time.Parse(time.RFC3339, mtime)
if err != nil {
return opts, InvalidArgument{
Bucket: bucket,
Object: object,
Err: fmt.Errorf("Unable to parse %s, failed with %w", xhttp.MinIOSourceMTime, err),
}
}
} else {
opts.MTime = UTCNow()
}
return opts, nil
}
@ -217,7 +160,7 @@ func putOpts(ctx context.Context, r *http.Request, bucket, object string, metada
}
}
mtimeStr := strings.TrimSpace(r.Header.Get(xhttp.MinIOSourceMTime))
mtime := UTCNow()
var mtime time.Time
if mtimeStr != "" {
mtime, err = time.Parse(time.RFC3339, mtimeStr)
if err != nil {
@ -227,6 +170,8 @@ func putOpts(ctx context.Context, r *http.Request, bucket, object string, metada
Err: fmt.Errorf("Unable to parse %s, failed with %w", xhttp.MinIOSourceMTime, err),
}
}
} else {
mtime = UTCNow()
}
etag := strings.TrimSpace(r.Header.Get(xhttp.MinIOSourceETag))
if etag != "" {

40
cmd/object-handlers.go
Unescape View File

@ -2705,23 +2705,6 @@ func (api objectAPIHandlers) DeleteObjectHandler(w http.ResponseWriter, r *http.
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
return
}
_, replicateDel := checkReplicateDelete(ctx, getObjectInfo, bucket, ObjectToDelete{ObjectName: object, VersionID: opts.VersionID})
if replicateDel {
if opts.VersionID != "" {
opts.VersionPurgeStatus = Pending
} else {
opts.DeleteMarkerReplicationStatus = string(replication.Pending)
}
}
if r.Header.Get(xhttp.AmzBucketReplicationStatus) == replication.Replica.String() {
// check if replica has permission to be deleted.
if apiErrCode := checkRequestAuthType(ctx, r, policy.ReplicateDeleteAction, bucket, object); apiErrCode != ErrNone {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
return
}
opts.DeleteMarkerReplicationStatus = replication.Replica.String()
}
apiErr := ErrNone
if rcfg, _ := globalBucketObjectLockSys.Get(bucket); rcfg.LockEnabled {
@ -2753,29 +2736,8 @@ func (api objectAPIHandlers) DeleteObjectHandler(w http.ResponseWriter, r *http.
}
// Ignore delete object errors while replying to client, since we are suppposed to reply only 204.
}
if replicateDel {
dmVersionID := ""
versionID := ""
if objInfo.DeleteMarker {
dmVersionID = objInfo.VersionID
} else {
versionID = objInfo.VersionID
}
globalReplicationState.queueReplicaDeleteTask(DeletedObjectVersionInfo{
DeletedObject: DeletedObject{
ObjectName: object,
VersionID: versionID,
DeleteMarkerVersionID: dmVersionID,
DeleteMarkerReplicationStatus: string(objInfo.ReplicationStatus),
DeleteMarkerMTime: objInfo.ModTime,
DeleteMarker: objInfo.DeleteMarker,
VersionPurgeStatus: objInfo.VersionPurgeStatus,
},
Bucket: bucket,
})
}
setPutObjHeaders(w, objInfo, true)
writeSuccessNoContent(w)
}

33
cmd/storage-datatypes.go
Unescape View File

@ -136,39 +136,6 @@ type FileInfo struct {
// Erasure info for all objects.
Erasure ErasureInfo
// DeleteMarkerReplicationStatus is set when this FileInfo represents
// replication on a DeleteMarker
MarkDeleted bool // mark this version as deleted
DeleteMarkerReplicationStatus string
VersionPurgeStatus VersionPurgeStatusType
}
// VersionPurgeStatusKey denotes purge status in metadata
const VersionPurgeStatusKey = "purgestatus"
// VersionPurgeStatusType represents status of a versioned delete or permanent delete w.r.t bucket replication
type VersionPurgeStatusType string
const (
// Pending - versioned delete replication is pending.
Pending VersionPurgeStatusType = "PENDING"
// Complete - versioned delete replication is now complete, erase version on disk.
Complete VersionPurgeStatusType = "COMPLETE"
// Failed - versioned delete replication failed.
Failed VersionPurgeStatusType = "FAILED"
)
// Empty returns true if purge status was not set.
func (v VersionPurgeStatusType) Empty() bool {
return string(v) == ""
}
// Pending returns true if the version is pending purge.
func (v VersionPurgeStatusType) Pending() bool {
return v == Pending || v == Failed
}
// newFileInfo - initializes new FileInfo, allocates a fresh erasure info.

128
cmd/storage-datatypes_gen.go
Unescape View File

@ -342,8 +342,8 @@ func (z *FileInfo) DecodeMsg(dc *msgp.Reader) (err error) {
err = msgp.WrapError(err)
return
}
if zb0001 != 16 {
err = msgp.ArrayError{Wanted: 16, Got: zb0001}
if zb0001 != 13 {
err = msgp.ArrayError{Wanted: 13, Got: zb0001}
return
}
z.Volume, err = dc.ReadString()
@ -448,32 +448,13 @@ func (z *FileInfo) DecodeMsg(dc *msgp.Reader) (err error) {
err = msgp.WrapError(err, "Erasure")
return
}
z.MarkDeleted, err = dc.ReadBool()
if err != nil {
err = msgp.WrapError(err, "MarkDeleted")
return
}
z.DeleteMarkerReplicationStatus, err = dc.ReadString()
if err != nil {
err = msgp.WrapError(err, "DeleteMarkerReplicationStatus")
return
}
{
var zb0004 string
zb0004, err = dc.ReadString()
if err != nil {
err = msgp.WrapError(err, "VersionPurgeStatus")
return
}
z.VersionPurgeStatus = VersionPurgeStatusType(zb0004)
}
return
}
// EncodeMsg implements msgp.Encodable
func (z *FileInfo) EncodeMsg(en *msgp.Writer) (err error) {
// array header, size 16
err = en.Append(0xdc, 0x0, 0x10)
// array header, size 13
err = en.Append(0x9d)
if err != nil {
return
}
@ -561,29 +542,14 @@ func (z *FileInfo) EncodeMsg(en *msgp.Writer) (err error) {
err = msgp.WrapError(err, "Erasure")
return
}
err = en.WriteBool(z.MarkDeleted)
if err != nil {
err = msgp.WrapError(err, "MarkDeleted")
return
}
err = en.WriteString(z.DeleteMarkerReplicationStatus)
if err != nil {
err = msgp.WrapError(err, "DeleteMarkerReplicationStatus")
return
}
err = en.WriteString(string(z.VersionPurgeStatus))
if err != nil {
err = msgp.WrapError(err, "VersionPurgeStatus")
return
}
return
}
// MarshalMsg implements msgp.Marshaler
func (z *FileInfo) MarshalMsg(b []byte) (o []byte, err error) {
o = msgp.Require(b, z.Msgsize())
// array header, size 16
o = append(o, 0xdc, 0x0, 0x10)
// array header, size 13
o = append(o, 0x9d)
o = msgp.AppendString(o, z.Volume)
o = msgp.AppendString(o, z.Name)
o = msgp.AppendString(o, z.VersionID)
@ -612,9 +578,6 @@ func (z *FileInfo) MarshalMsg(b []byte) (o []byte, err error) {
err = msgp.WrapError(err, "Erasure")
return
}
o = msgp.AppendBool(o, z.MarkDeleted)
o = msgp.AppendString(o, z.DeleteMarkerReplicationStatus)
o = msgp.AppendString(o, string(z.VersionPurgeStatus))
return
}
@ -626,8 +589,8 @@ func (z *FileInfo) UnmarshalMsg(bts []byte) (o []byte, err error) {
err = msgp.WrapError(err)
return
}
if zb0001 != 16 {
err = msgp.ArrayError{Wanted: 16, Got: zb0001}
if zb0001 != 13 {
err = msgp.ArrayError{Wanted: 13, Got: zb0001}
return
}
z.Volume, bts, err = msgp.ReadStringBytes(bts)
@ -732,32 +695,13 @@ func (z *FileInfo) UnmarshalMsg(bts []byte) (o []byte, err error) {
err = msgp.WrapError(err, "Erasure")
return
}
z.MarkDeleted, bts, err = msgp.ReadBoolBytes(bts)
if err != nil {
err = msgp.WrapError(err, "MarkDeleted")
return
}
z.DeleteMarkerReplicationStatus, bts, err = msgp.ReadStringBytes(bts)
if err != nil {
err = msgp.WrapError(err, "DeleteMarkerReplicationStatus")
return
}
{
var zb0004 string
zb0004, bts, err = msgp.ReadStringBytes(bts)
if err != nil {
err = msgp.WrapError(err, "VersionPurgeStatus")
return
}
z.VersionPurgeStatus = VersionPurgeStatusType(zb0004)
}
o = bts
return
}
// Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message
func (z *FileInfo) Msgsize() (s int) {
s = 3 + msgp.StringPrefixSize + len(z.Volume) + msgp.StringPrefixSize + len(z.Name) + msgp.StringPrefixSize + len(z.VersionID) + msgp.BoolSize + msgp.BoolSize + msgp.StringPrefixSize + len(z.DataDir) + msgp.BoolSize + msgp.TimeSize + msgp.Int64Size + msgp.Uint32Size + msgp.MapHeaderSize
s = 1 + msgp.StringPrefixSize + len(z.Volume) + msgp.StringPrefixSize + len(z.Name) + msgp.StringPrefixSize + len(z.VersionID) + msgp.BoolSize + msgp.BoolSize + msgp.StringPrefixSize + len(z.DataDir) + msgp.BoolSize + msgp.TimeSize + msgp.Int64Size + msgp.Uint32Size + msgp.MapHeaderSize
if z.Metadata != nil {
for za0001, za0002 := range z.Metadata {
_ = za0002
@ -768,7 +712,7 @@ func (z *FileInfo) Msgsize() (s int) {
for za0003 := range z.Parts {
s += z.Parts[za0003].Msgsize()
}
s += z.Erasure.Msgsize() + msgp.BoolSize + msgp.StringPrefixSize + len(z.DeleteMarkerReplicationStatus) + msgp.StringPrefixSize + len(string(z.VersionPurgeStatus))
s += z.Erasure.Msgsize()
return
}
@ -1337,58 +1281,6 @@ func (z *FilesInfoVersions) Msgsize() (s int) {
return
}
// DecodeMsg implements msgp.Decodable
func (z *VersionPurgeStatusType) DecodeMsg(dc *msgp.Reader) (err error) {
{
var zb0001 string
zb0001, err = dc.ReadString()
if err != nil {
err = msgp.WrapError(err)
return
}
(*z) = VersionPurgeStatusType(zb0001)
}
return
}
// EncodeMsg implements msgp.Encodable
func (z VersionPurgeStatusType) EncodeMsg(en *msgp.Writer) (err error) {
err = en.WriteString(string(z))
if err != nil {
err = msgp.WrapError(err)
return
}
return
}
// MarshalMsg implements msgp.Marshaler
func (z VersionPurgeStatusType) MarshalMsg(b []byte) (o []byte, err error) {
o = msgp.Require(b, z.Msgsize())
o = msgp.AppendString(o, string(z))
return
}
// UnmarshalMsg implements msgp.Unmarshaler
func (z *VersionPurgeStatusType) UnmarshalMsg(bts []byte) (o []byte, err error) {
{
var zb0001 string
zb0001, bts, err = msgp.ReadStringBytes(bts)
if err != nil {
err = msgp.WrapError(err)
return
}
(*z) = VersionPurgeStatusType(zb0001)
}
o = bts
return
}
// Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message
func (z VersionPurgeStatusType) Msgsize() (s int) {
s = msgp.StringPrefixSize + len(string(z))
return
}
// DecodeMsg implements msgp.Decodable
func (z *VolInfo) DecodeMsg(dc *msgp.Reader) (err error) {
var field []byte

74
cmd/web-handlers.go
Unescape View File

@ -612,10 +612,6 @@ func (web *webAPIHandlers) RemoveObject(r *http.Request, args *RemoveObjectArgs,
if web.CacheAPI() != nil {
deleteObjects = web.CacheAPI().DeleteObjects
}
getObjectInfoFn := objectAPI.GetObjectInfo
if web.CacheAPI() != nil {
getObjectInfoFn = web.CacheAPI().GetObjectInfo
}
claims, owner, authErr := webRequestAuthenticate(r)
if authErr != nil {
@ -718,27 +714,8 @@ next:
return toJSONError(ctx, errAccessDenied)
}
}
_, replicateDel := checkReplicateDelete(ctx, getObjectInfoFn, args.BucketName, ObjectToDelete{ObjectName: objectName})
if replicateDel {
opts.DeleteMarkerReplicationStatus = string(replication.Pending)
opts.DeleteMarker = true
}
oi, err := deleteObject(ctx, objectAPI, web.CacheAPI(), args.BucketName, objectName, nil, r, opts)
if replicateDel {
globalReplicationState.queueReplicaDeleteTask(DeletedObjectVersionInfo{
DeletedObject: DeletedObject{
ObjectName: objectName,
DeleteMarkerVersionID: oi.VersionID,
DeleteMarkerReplicationStatus: string(oi.ReplicationStatus),
DeleteMarkerMTime: oi.ModTime,
DeleteMarker: oi.DeleteMarker,
VersionPurgeStatus: oi.VersionPurgeStatus,
},
Bucket: args.BucketName,
})
}
_, err = deleteObject(ctx, objectAPI, web.CacheAPI(), args.BucketName, objectName, nil, r, opts)
logger.LogIf(ctx, err)
continue
}
@ -783,40 +760,9 @@ next:
// Reached maximum delete requests, attempt a delete for now.
break
}
if obj.ReplicationStatus == replication.Replica {
if authErr == errNoAuthToken {
// Check if object is allowed to be deleted anonymously
if !globalPolicySys.IsAllowed(policy.Args{
Action: iampolicy.ReplicateDeleteAction,
BucketName: args.BucketName,
ConditionValues: getConditionValues(r, "", "", nil),
IsOwner: false,
ObjectName: objectName,
}) {
return toJSONError(ctx, errAccessDenied)
}
} else {
if !globalIAMSys.IsAllowed(iampolicy.Args{
AccountName: claims.AccessKey,
Action: iampolicy.ReplicateDeleteAction,
BucketName: args.BucketName,
ConditionValues: getConditionValues(r, "", claims.AccessKey, claims.Map()),
IsOwner: owner,
ObjectName: objectName,
Claims: claims.Map(),
}) {
return toJSONError(ctx, errAccessDenied)
}
}
}
// since versioned delete is not available on web browser, yet - this is a simple DeleteMarker replication
_, replicateDel := checkReplicateDelete(ctx, getObjectInfoFn, args.BucketName, ObjectToDelete{ObjectName: obj.Name})
objToDel := ObjectToDelete{ObjectName: obj.Name}
if replicateDel {
objToDel.DeleteMarkerReplicationStatus = string(replication.Pending)
}
objects = append(objects, objToDel)
objects = append(objects, ObjectToDelete{
ObjectName: obj.Name,
})
}
// Nothing to do.
@ -826,11 +772,7 @@ next:
// Deletes a list of objects.
deletedObjects, errs := deleteObjects(ctx, args.BucketName, objects, opts)
for i, err := range errs {
if err != nil && !isErrObjectNotFound(err) {
deletedObjects[i].DeleteMarkerReplicationStatus = objects[i].DeleteMarkerReplicationStatus
deletedObjects[i].VersionPurgeStatus = objects[i].VersionPurgeStatus
}
for _, err := range errs {
if err != nil {
logger.LogIf(ctx, err)
break next
@ -857,12 +799,6 @@ next:
UserAgent: r.UserAgent(),
Host: handlers.GetSourceIP(r),
})
if dobj.DeleteMarkerReplicationStatus == string(replication.Pending) || dobj.VersionPurgeStatus == Pending {
globalReplicationState.queueReplicaDeleteTask(DeletedObjectVersionInfo{
DeletedObject: dobj,
Bucket: args.BucketName,
})
}
}
}
}

4
cmd/xl-storage-format-utils.go
Unescape View File

@ -34,13 +34,13 @@ func (v versionsSorter) Less(i, j int) bool {
return v[i].ModTime.After(v[j].ModTime)
}
func getFileInfoVersions(xlMetaBuf []byte, volume, path string, showPendingDeletes bool) (FileInfoVersions, error) {
func getFileInfoVersions(xlMetaBuf []byte, volume, path string) (FileInfoVersions, error) {
if isXL2V1Format(xlMetaBuf) {
var xlMeta xlMetaV2
if err := xlMeta.Load(xlMetaBuf); err != nil {
return FileInfoVersions{}, err
}
versions, latestModTime, err := xlMeta.ListVersions(volume, path, showPendingDeletes)
versions, latestModTime, err := xlMeta.ListVersions(volume, path)
if err != nil {
return FileInfoVersions{}, err
}

74
cmd/xl-storage-format-v2.go
Unescape View File

@ -139,9 +139,8 @@ func (e ChecksumAlgo) valid() bool {
// xlMetaV2DeleteMarker defines the data struct for the delete marker journal type
type xlMetaV2DeleteMarker struct {
VersionID [16]byte `json:"ID" msg:"ID"` // Version ID for delete marker
ModTime int64 `json:"MTime" msg:"MTime"` // Object delete marker modified time
MetaSys map[string][]byte `json:"MetaSys,omitempty" msg:"MetaSys,omitempty"` // Delete marker internal metadata
VersionID [16]byte `json:"ID" msg:"ID"` // Version ID for delete marker
ModTime int64 `json:"MTime" msg:"MTime"` // Object delete marker modified time
}
// xlMetaV2Object defines the data struct for object journal type
@ -355,13 +354,11 @@ func (j xlMetaV2DeleteMarker) ToFileInfo(volume, path string) (FileInfo, error)
versionID = uuid.UUID(j.VersionID).String()
}
fi := FileInfo{
Volume: volume,
Name: path,
ModTime: time.Unix(0, j.ModTime).UTC(),
VersionID: versionID,
Deleted: true,
DeleteMarkerReplicationStatus: string(j.MetaSys[xhttp.AmzBucketReplicationStatus]),
VersionPurgeStatus: VersionPurgeStatusType(string(j.MetaSys[VersionPurgeStatusKey])),
Volume: volume,
Name: path,
ModTime: time.Unix(0, j.ModTime).UTC(),
VersionID: versionID,
Deleted: true,
}
return fi, nil
}
@ -411,9 +408,6 @@ func (j xlMetaV2Object) ToFileInfo(volume, path string) (FileInfo, error) {
if strings.HasPrefix(strings.ToLower(k), ReservedMetadataPrefixLower) {
fi.Metadata[k] = string(v)
}
if strings.EqualFold(k, VersionPurgeStatusKey) {
fi.VersionPurgeStatus = VersionPurgeStatusType(string(v))
}
}
fi.Erasure.Algorithm = j.ErasureAlgorithm.String()
fi.Erasure.Index = j.ErasureIndex
@ -452,36 +446,12 @@ func (z *xlMetaV2) DeleteVersion(fi FileInfo) (string, bool, error) {
DeleteMarker: &xlMetaV2DeleteMarker{
VersionID: uv,
ModTime: fi.ModTime.UnixNano(),
MetaSys: make(map[string][]byte),
},
}
if !ventry.Valid() {
return "", false, errors.New("internal error: invalid version entry generated")
}
}
updateVersion := false
if fi.VersionPurgeStatus.Empty() && (fi.DeleteMarkerReplicationStatus == "REPLICA" || fi.DeleteMarkerReplicationStatus == "") {
updateVersion = fi.MarkDeleted
} else {
// for replication scenario
if fi.Deleted && fi.VersionPurgeStatus != Complete {
if !fi.VersionPurgeStatus.Empty() || fi.DeleteMarkerReplicationStatus != "" {
updateVersion = true
}
}
// object or delete-marker versioned delete is not complete
if !fi.VersionPurgeStatus.Empty() && fi.VersionPurgeStatus != Complete {
updateVersion = true
}
}
if fi.Deleted {
if fi.DeleteMarkerReplicationStatus != "" {
ventry.DeleteMarker.MetaSys[xhttp.AmzBucketReplicationStatus] = []byte(fi.DeleteMarkerReplicationStatus)
}
if !fi.VersionPurgeStatus.Empty() {
ventry.DeleteMarker.MetaSys[VersionPurgeStatusKey] = []byte(fi.VersionPurgeStatus)
}
}
for i, version := range z.Versions {
if !version.Valid() {
@ -498,28 +468,12 @@ func (z *xlMetaV2) DeleteVersion(fi FileInfo) (string, bool, error) {
}
case DeleteType:
if bytes.Equal(version.DeleteMarker.VersionID[:], uv[:]) {
if updateVersion {
delete(z.Versions[i].DeleteMarker.MetaSys, xhttp.AmzBucketReplicationStatus)
delete(z.Versions[i].DeleteMarker.MetaSys, VersionPurgeStatusKey)
if fi.DeleteMarkerReplicationStatus != "" {
z.Versions[i].DeleteMarker.MetaSys[xhttp.AmzBucketReplicationStatus] = []byte(fi.DeleteMarkerReplicationStatus)
}
if !fi.VersionPurgeStatus.Empty() {
z.Versions[i].DeleteMarker.MetaSys[VersionPurgeStatusKey] = []byte(fi.VersionPurgeStatus)
}
} else {
z.Versions = append(z.Versions[:i], z.Versions[i+1:]...)
if fi.MarkDeleted && (fi.VersionPurgeStatus.Empty() || (fi.VersionPurgeStatus != Complete)) {
z.Versions = append(z.Versions, ventry)
}
z.Versions = append(z.Versions[:i], z.Versions[i+1:]...)
if fi.Deleted {
z.Versions = append(z.Versions, ventry)
}
return "", len(z.Versions) == 0, nil
}
case ObjectType:
if bytes.Equal(version.ObjectV2.VersionID[:], uv[:]) && updateVersion {
z.Versions[i].ObjectV2.MetaSys[VersionPurgeStatusKey] = []byte(fi.VersionPurgeStatus)
return "", len(z.Versions) == 0, nil
}
}
}
@ -564,6 +518,7 @@ func (z *xlMetaV2) DeleteVersion(fi FileInfo) (string, bool, error) {
z.Versions = append(z.Versions, ventry)
return "", false, nil
}
return "", false, errFileVersionNotFound
}
@ -583,9 +538,7 @@ func (z xlMetaV2) TotalSize() int64 {
// ListVersions lists current versions, and current deleted
// versions returns error for unexpected entries.
// showPendingDeletes is set to true if ListVersions needs to list objects marked deleted
// but waiting to be replicated
func (z xlMetaV2) ListVersions(volume, path string, showPendingDeletes bool) (versions []FileInfo, modTime time.Time, err error) {
func (z xlMetaV2) ListVersions(volume, path string) (versions []FileInfo, modTime time.Time, err error) {
var latestModTime time.Time
var latestVersionID string
for _, version := range z.Versions {
@ -598,9 +551,6 @@ func (z xlMetaV2) ListVersions(volume, path string, showPendingDeletes bool) (ve
fi, err = version.ObjectV2.ToFileInfo(volume, path)
case DeleteType:
fi, err = version.DeleteMarker.ToFileInfo(volume, path)
if !fi.VersionPurgeStatus.Empty() && !showPendingDeletes {
continue
}
case LegacyType:
fi, err = version.ObjectV1.ToFileInfo(volume, path)
}

230
cmd/xl-storage-format-v2_gen.go
Unescape View File

@ -338,36 +338,6 @@ func (z *xlMetaV2DeleteMarker) DecodeMsg(dc *msgp.Reader) (err error) {
err = msgp.WrapError(err, "ModTime")
return
}
case "MetaSys":
var zb0002 uint32
zb0002, err = dc.ReadMapHeader()
if err != nil {
err = msgp.WrapError(err, "MetaSys")
return
}
if z.MetaSys == nil {
z.MetaSys = make(map[string][]byte, zb0002)
} else if len(z.MetaSys) > 0 {
for key := range z.MetaSys {
delete(z.MetaSys, key)
}
}
for zb0002 > 0 {
zb0002--
var za0002 string
var za0003 []byte
za0002, err = dc.ReadString()
if err != nil {
err = msgp.WrapError(err, "MetaSys")
return
}
za0003, err = dc.ReadBytes(za0003)
if err != nil {
err = msgp.WrapError(err, "MetaSys", za0002)
return
}
z.MetaSys[za0002] = za0003
}
default:
err = dc.Skip()
if err != nil {
@ -381,23 +351,9 @@ func (z *xlMetaV2DeleteMarker) DecodeMsg(dc *msgp.Reader) (err error) {
// EncodeMsg implements msgp.Encodable
func (z *xlMetaV2DeleteMarker) EncodeMsg(en *msgp.Writer) (err error) {
// omitempty: check for empty values
zb0001Len := uint32(3)
var zb0001Mask uint8 /* 3 bits */
if z.MetaSys == nil {
zb0001Len--
zb0001Mask |= 0x4
}
// variable map header, size zb0001Len
err = en.Append(0x80 | uint8(zb0001Len))
if err != nil {
return
}
if zb0001Len == 0 {
return
}
// map header, size 2
// write "ID"
err = en.Append(0xa2, 0x49, 0x44)
err = en.Append(0x82, 0xa2, 0x49, 0x44)
if err != nil {
return
}
@ -416,63 +372,19 @@ func (z *xlMetaV2DeleteMarker) EncodeMsg(en *msgp.Writer) (err error) {
err = msgp.WrapError(err, "ModTime")
return
}
if (zb0001Mask & 0x4) == 0 { // if not empty
// write "MetaSys"
err = en.Append(0xa7, 0x4d, 0x65, 0x74, 0x61, 0x53, 0x79, 0x73)
if err != nil {
return
}
err = en.WriteMapHeader(uint32(len(z.MetaSys)))
if err != nil {
err = msgp.WrapError(err, "MetaSys")
return
}
for za0002, za0003 := range z.MetaSys {
err = en.WriteString(za0002)
if err != nil {
err = msgp.WrapError(err, "MetaSys")
return
}
err = en.WriteBytes(za0003)
if err != nil {
err = msgp.WrapError(err, "MetaSys", za0002)
return
}
}
}
return
}
// MarshalMsg implements msgp.Marshaler
func (z *xlMetaV2DeleteMarker) MarshalMsg(b []byte) (o []byte, err error) {
o = msgp.Require(b, z.Msgsize())
// omitempty: check for empty values
zb0001Len := uint32(3)
var zb0001Mask uint8 /* 3 bits */
if z.MetaSys == nil {
zb0001Len--
zb0001Mask |= 0x4
}
// variable map header, size zb0001Len
o = append(o, 0x80|uint8(zb0001Len))
if zb0001Len == 0 {
return
}
// map header, size 2
// string "ID"
o = append(o, 0xa2, 0x49, 0x44)
o = append(o, 0x82, 0xa2, 0x49, 0x44)
o = msgp.AppendBytes(o, (z.VersionID)[:])
// string "MTime"
o = append(o, 0xa5, 0x4d, 0x54, 0x69, 0x6d, 0x65)
o = msgp.AppendInt64(o, z.ModTime)
if (zb0001Mask & 0x4) == 0 { // if not empty
// string "MetaSys"
o = append(o, 0xa7, 0x4d, 0x65, 0x74, 0x61, 0x53, 0x79, 0x73)
o = msgp.AppendMapHeader(o, uint32(len(z.MetaSys)))
for za0002, za0003 := range z.MetaSys {
o = msgp.AppendString(o, za0002)
o = msgp.AppendBytes(o, za0003)
}
}
return
}
@ -506,36 +418,6 @@ func (z *xlMetaV2DeleteMarker) UnmarshalMsg(bts []byte) (o []byte, err error) {
err = msgp.WrapError(err, "ModTime")
return
}
case "MetaSys":
var zb0002 uint32
zb0002, bts, err = msgp.ReadMapHeaderBytes(bts)
if err != nil {
err = msgp.WrapError(err, "MetaSys")
return
}
if z.MetaSys == nil {
z.MetaSys = make(map[string][]byte, zb0002)
} else if len(z.MetaSys) > 0 {
for key := range z.MetaSys {
delete(z.MetaSys, key)
}
}
for zb0002 > 0 {
var za0002 string
var za0003 []byte
zb0002--
za0002, bts, err = msgp.ReadStringBytes(bts)
if err != nil {
err = msgp.WrapError(err, "MetaSys")
return
}
za0003, bts, err = msgp.ReadBytesBytes(bts, za0003)
if err != nil {
err = msgp.WrapError(err, "MetaSys", za0002)
return
}
z.MetaSys[za0002] = za0003
}
default:
bts, err = msgp.Skip(bts)
if err != nil {
@ -550,13 +432,7 @@ func (z *xlMetaV2DeleteMarker) UnmarshalMsg(bts []byte) (o []byte, err error) {
// Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message
func (z *xlMetaV2DeleteMarker) Msgsize() (s int) {
s = 1 + 3 + msgp.ArrayHeaderSize + (16 * (msgp.ByteSize)) + 6 + msgp.Int64Size + 8 + msgp.MapHeaderSize
if z.MetaSys != nil {
for za0002, za0003 := range z.MetaSys {
_ = za0003
s += msgp.StringPrefixSize + len(za0002) + msgp.BytesPrefixSize + len(za0003)
}
}
s = 1 + 3 + msgp.ArrayHeaderSize + (16 * (msgp.ByteSize)) + 6 + msgp.Int64Size
return
}
@ -1533,11 +1409,40 @@ func (z *xlMetaV2Version) DecodeMsg(dc *msgp.Reader) (err error) {
if z.DeleteMarker == nil {
z.DeleteMarker = new(xlMetaV2DeleteMarker)
}
err = z.DeleteMarker.DecodeMsg(dc)
var zb0003 uint32
zb0003, err = dc.ReadMapHeader()
if err != nil {
err = msgp.WrapError(err, "DeleteMarker")
return
}
for zb0003 > 0 {
zb0003--
field, err = dc.ReadMapKeyPtr()
if err != nil {
err = msgp.WrapError(err, "DeleteMarker")
return
}
switch msgp.UnsafeString(field) {
case "ID":
err = dc.ReadExactBytes((z.DeleteMarker.VersionID)[:])
if err != nil {
err = msgp.WrapError(err, "DeleteMarker", "VersionID")
return
}
case "MTime":
z.DeleteMarker.ModTime, err = dc.ReadInt64()
if err != nil {
err = msgp.WrapError(err, "DeleteMarker", "ModTime")
return
}
default:
err = dc.Skip()
if err != nil {
err = msgp.WrapError(err, "DeleteMarker")
return
}
}
}
}
default:
err = dc.Skip()
@ -1635,9 +1540,25 @@ func (z *xlMetaV2Version) EncodeMsg(en *msgp.Writer) (err error) {
return
}
} else {
err = z.DeleteMarker.EncodeMsg(en)
// map header, size 2
// write "ID"
err = en.Append(0x82, 0xa2, 0x49, 0x44)
if err != nil {
err = msgp.WrapError(err, "DeleteMarker")
return
}
err = en.WriteBytes((z.DeleteMarker.VersionID)[:])
if err != nil {
err = msgp.WrapError(err, "DeleteMarker", "VersionID")
return
}
// write "MTime"
err = en.Append(0xa5, 0x4d, 0x54, 0x69, 0x6d, 0x65)
if err != nil {
return
}
err = en.WriteInt64(z.DeleteMarker.ModTime)
if err != nil {
err = msgp.WrapError(err, "DeleteMarker", "ModTime")
return
}
}
@ -1703,11 +1624,13 @@ func (z *xlMetaV2Version) MarshalMsg(b []byte) (o []byte, err error) {
if z.DeleteMarker == nil {
o = msgp.AppendNil(o)
} else {
o, err = z.DeleteMarker.MarshalMsg(o)
if err != nil {
err = msgp.WrapError(err, "DeleteMarker")
return
}
// map header, size 2
// string "ID"
o = append(o, 0x82, 0xa2, 0x49, 0x44)
o = msgp.AppendBytes(o, (z.DeleteMarker.VersionID)[:])
// string "MTime"
o = append(o, 0xa5, 0x4d, 0x54, 0x69, 0x6d, 0x65)
o = msgp.AppendInt64(o, z.DeleteMarker.ModTime)
}
}
return
@ -1786,11 +1709,40 @@ func (z *xlMetaV2Version) UnmarshalMsg(bts []byte) (o []byte, err error) {
if z.DeleteMarker == nil {
z.DeleteMarker = new(xlMetaV2DeleteMarker)
}
bts, err = z.DeleteMarker.UnmarshalMsg(bts)
var zb0003 uint32
zb0003, bts, err = msgp.ReadMapHeaderBytes(bts)
if err != nil {
err = msgp.WrapError(err, "DeleteMarker")
return
}
for zb0003 > 0 {
zb0003--
field, bts, err = msgp.ReadMapKeyZC(bts)
if err != nil {
err = msgp.WrapError(err, "DeleteMarker")
return
}
switch msgp.UnsafeString(field) {
case "ID":
bts, err = msgp.ReadExactBytes(bts, (z.DeleteMarker.VersionID)[:])
if err != nil {
err = msgp.WrapError(err, "DeleteMarker", "VersionID")
return
}
case "MTime":
z.DeleteMarker.ModTime, bts, err = msgp.ReadInt64Bytes(bts)
if err != nil {
err = msgp.WrapError(err, "DeleteMarker", "ModTime")
return
}
default:
bts, err = msgp.Skip(bts)
if err != nil {
err = msgp.WrapError(err, "DeleteMarker")
return
}
}
}
}
default:
bts, err = msgp.Skip(bts)
@ -1822,7 +1774,7 @@ func (z *xlMetaV2Version) Msgsize() (s int) {
if z.DeleteMarker == nil {
s += msgp.NilSize
} else {
s += z.DeleteMarker.Msgsize()
s += 1 + 3 + msgp.ArrayHeaderSize + (16 * (msgp.ByteSize)) + 6 + msgp.Int64Size
}
return
}

6
cmd/xl-storage.go
Unescape View File

@ -379,7 +379,7 @@ func (s *xlStorage) CrawlAndGetDataUsage(ctx context.Context, cache dataUsageCac
// Remove filename which is the meta file.
item.transformMetaDir()
fivs, err := getFileInfoVersions(buf, item.bucket, item.objectPath(), true)
fivs, err := getFileInfoVersions(buf, item.bucket, item.objectPath())
if err != nil {
return 0, errSkipFile
}
@ -423,7 +423,7 @@ func (s *xlStorage) CrawlAndGetDataUsage(ctx context.Context, cache dataUsageCac
}
totalSize += size
}
item.healReplication(ctx, objAPI, actionMeta{oi: version.ToObjectInfo(item.bucket, item.objectPath())})
item.healReplication(ctx, objAPI, actionMeta{oi: oi})
}
return totalSize, nil
})
@ -897,7 +897,7 @@ func (s *xlStorage) WalkVersions(ctx context.Context, volume, dirPath, marker st
continue
}
fiv, err = getFileInfoVersions(xlMetaBuf, volume, walkResult.entry, false)
fiv, err = getFileInfoVersions(xlMetaBuf, volume, walkResult.entry)
if err != nil {
continue
}

BIN
docs/bucket/replication/DELETE_bucket-replication.png
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 37 KiB

29
docs/bucket/replication/README.md
Unescape View File

@ -88,7 +88,6 @@ The replication configuration can now be added to the source bucket by applying
"Status": "Enabled",
"Priority": 1,
"DeleteMarkerReplication": { "Status": "Disabled" },
"DeleteReplication": { "Status": "Disabled" },
"Filter" : {
"And": {
"Prefix": "Tax",
@ -113,8 +112,18 @@ The replication configuration can now be added to the source bucket by applying
}
```
```
mc replicate add myminio/srcbucket/Tax --priority 1 --arn "arn:minio:replication:us-east-1:c5be6b16-769d-432a-9ef1-4567081f3566:destbucket" --tags "Year=2019&Company=AcmeCorp" --storage-class "STANDARD" --remote-bucket "destbucket"
Replication configuration applied successfully to myminio/srcbucket.
```
The replication configuration follows [AWS S3 Spec](https://docs.aws.amazon.com/AmazonS3/latest/dev/replication-add-config.html). Any objects uploaded to the source bucket that meet replication criteria will now be automatically replicated by the MinIO server to the remote destination bucket. Replication can be disabled at any time by disabling specific rules in the configuration or deleting the replication configuration entirely.
When an object is deleted from the source bucket, the replica will not be deleted as per S3 spec.
![delete](https://raw.githubusercontent.com/minio/minio/master/docs/bucket/replication/DELETE_bucket_replication.png)
When object locking is used in conjunction with replication, both source and destination buckets needs to have object locking enabled. Similarly objects encrypted on the server side, will be replicated if destination also supports encryption.
Replication status can be seen in the metadata on the source and destination objects. On the source side, the `X-Amz-Replication-Status` changes from `PENDING` to `COMPLETE` or `FAILED` after replication attempt either succeeded or failed respectively. On the destination side, a `X-Amz-Replication-Status` status of `REPLICA` indicates that the object was replicated successfully. Any replication failures are automatically re-attempted during a periodic disk crawl cycle.
@ -127,24 +136,6 @@ It is recommended that replication be run in a system with atleast two CPU's ava
![head](https://raw.githubusercontent.com/minio/minio/master/docs/bucket/replication/HEAD_bucket_replication.png)
## MinIO Extension
Delete marker replication is allowed in [AWS V1 Configuration](https://aws.amazon.com/blogs/storage/managing-delete-marker-replication-in-amazon-s3/) but not in V2 configuration. The MinIO implementation above is based on V2 configuration, however it has been extended to allow both DeleteMarker replication and replication of versioned deletes with the `DeleteMarkerReplication` and `DeleteReplication` fields in the replication configuration above. By default, this is set to `Disabled` unless the user specifies it while adding a replication rule.
When an object is deleted from the source bucket, the corresponding replica version will be marked deleted if delete marker replication is enabled in the replication configuration.Replication of deletes that specify a version id (a.k.a hard deletes) can be enabled by setting the `DeleteReplication` status to enabled in the replication configuration. This is a MinIO specific extension that can be enabled using the `mc replicate add` or `mc replicate edit` command with the --replicate "delete" flag.
Note that due to this extension behavior, AWS SDK's may not support the extension functionality pertaining to replicating versioned deletes.
To add a replication rule allowing both delete marker replication, versioned delete replication or both specify the --replicate flag with comma separated values as in the example below.
Additional permission of "s3:ReplicateDelete" action would need to be specified on the access key configured for the target cluster if Delete Marker replication or versioned delete replication is enabled.
```
mc replicate add myminio/srcbucket/Tax --priority 1 --arn "arn:minio:replication:us-east-1:c5be6b16-769d-432a-9ef1-4567081f3566:destbucket" --tags "Year=2019&Company=AcmeCorp" --storage-class "STANDARD" --remote-bucket "destbucket" --replicate "delete,delete-marker"
Replication configuration applied successfully to myminio/srcbucket.
```
Note that both source and target instance need to be upgraded to latest release to take advantage of Delete marker replication.
![delete](https://raw.githubusercontent.com/minio/minio/master/docs/bucket/replication/DELETE_bucket_replication.png)
## Explore Further
- [MinIO Bucket Versioning Implementation](https://docs.minio.io/docs/minio-bucket-versioning-guide.html)
- [MinIO Client Quickstart Guide](https://docs.minio.io/docs/minio-client-quickstart-guide.html)

3
go.mod
Unescape View File

@ -48,7 +48,7 @@ require (
github.com/miekg/dns v1.1.8
github.com/minio/cli v1.22.0
github.com/minio/highwayhash v1.0.0
github.com/minio/minio-go/v7 v7.0.6-0.20201013215222-14baba9e61ac
github.com/minio/minio-go/v7 v7.0.6-0.20200929220449-755b5633803a
github.com/minio/selfupdate v0.3.1
github.com/minio/sha256-simd v0.1.1
github.com/minio/simdjson-go v0.1.5
@ -75,6 +75,7 @@ require (
github.com/tidwall/gjson v1.3.5
github.com/tidwall/sjson v1.0.4
github.com/tinylib/msgp v1.1.3
github.com/ttacon/chalk v0.0.0-20160626202418-22c06c80ed31 // indirect
github.com/valyala/tcplisten v0.0.0-20161114210144-ceec8f93295a
github.com/willf/bitset v1.1.11 // indirect
github.com/willf/bloom v2.0.3+incompatible

32
go.sum
Unescape View File

@ -46,7 +46,6 @@ github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da/go.mod h1:Q73ZrmV
github.com/armon/go-metrics v0.0.0-20190430140413-ec5e00d3c878 h1:EFSB7Zo9Eg91v7MJPVsifUysc/wPdN+NOnVe6bWbdBM=
github.com/armon/go-metrics v0.0.0-20190430140413-ec5e00d3c878/go.mod h1:3AMJUQhVx52RsWOnlkpikZr01T/yAVN2gn0861vByNg=
github.com/armon/go-radix v0.0.0-20180808171621-7fddfc383310/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgIH9cCH8=
github.com/aws/aws-sdk-go v1.29.11 h1:f1QJRPu30p0i1lzKhkSSaZFudFGCra2HKgdE442nN6c=
github.com/aws/aws-sdk-go v1.29.11/go.mod h1:1KvfttTE3SPKMpo8g2c6jL3ZKfXtFvKscTgahTma5Xg=
github.com/bcicen/jstream v0.0.0-20190220045926-16c1f8af81c2 h1:M+TYzBcNIRyzPRg66ndEqUMd7oWDmhvdQmaPC6EZNwM=
github.com/bcicen/jstream v0.0.0-20190220045926-16c1f8af81c2/go.mod h1:RDu/qcrnpEdJC/p8tx34+YBFqqX71lB7dOX9QE+ZC4M=
@ -236,7 +235,6 @@ github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANyt
github.com/jcmturner/gofork v0.0.0-20180107083740-2aebee971930/go.mod h1:MK8+TM0La+2rjBD4jE12Kj1pCCxK7d2LK/UM3ncEo0o=
github.com/jcmturner/gofork v0.0.0-20190328161633-dc7c13fece03 h1:FUwcHNlEqkqLjLBdCp5PRlCFijNjvcYANOZXzCfXwCM=
github.com/jcmturner/gofork v0.0.0-20190328161633-dc7c13fece03/go.mod h1:MK8+TM0La+2rjBD4jE12Kj1pCCxK7d2LK/UM3ncEo0o=
github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af h1:pmfjZENx5imkbgOkpRUYLnmbU7UEFbjtDA2hxJ1ichM=
github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af/go.mod h1:Nht3zPeWKUH0NzdCt2Blrr5ys8VGpn0CEB0cQHVjt7k=
github.com/jonboulle/clockwork v0.1.0 h1:VKV+ZcuP6l3yW9doeqz6ziZGgcynBVQO+obU0+0hcPo=
github.com/jonboulle/clockwork v0.1.0/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22gdxWY5EU2bo=
@ -310,8 +308,8 @@ github.com/minio/highwayhash v1.0.0 h1:iMSDhgUILCr0TNm8LWlSjF8N0ZIj2qbO8WHp6Q/J2
github.com/minio/highwayhash v1.0.0/go.mod h1:xQboMTeM9nY9v/LlAOxFctujiv5+Aq2hR5dxBpaMbdc=
github.com/minio/md5-simd v1.1.0 h1:QPfiOqlZH+Cj9teu0t9b1nTBfPbyTl16Of5MeuShdK4=
github.com/minio/md5-simd v1.1.0/go.mod h1:XpBqgZULrMYD3R+M28PcmP0CkI7PEMzB3U77ZrKZ0Gw=
github.com/minio/minio-go/v7 v7.0.6-0.20201013215222-14baba9e61ac h1:0meQIZTQR/JkAxfygReKcb15QINBKpFd4LII2PT5jSY=
github.com/minio/minio-go/v7 v7.0.6-0.20201013215222-14baba9e61ac/go.mod h1:CSt2ETZNs+bIIhWTse0mcZKZWMGrFU7Er7RR0TmkDYk=
github.com/minio/minio-go/v7 v7.0.6-0.20200929220449-755b5633803a h1:duFwQxzoPzBt4LbaftSVVTDwRrpuMIKI6XoJ5mKSaOs=
github.com/minio/minio-go/v7 v7.0.6-0.20200929220449-755b5633803a/go.mod h1:CSt2ETZNs+bIIhWTse0mcZKZWMGrFU7Er7RR0TmkDYk=
github.com/minio/selfupdate v0.3.1 h1:BWEFSNnrZVMUWXbXIgLDNDjbejkmpAmZvy/nCz1HlEs=
github.com/minio/selfupdate v0.3.1/go.mod h1:b8ThJzzH7u2MkF6PcIra7KaXO9Khf6alWPvMSyTDCFM=
github.com/minio/sha256-simd v0.1.1 h1:5QHSlgo3nt5yKOJrC7W8w7X+NFl8cMPZm96iu8kKUJU=
@ -332,9 +330,11 @@ github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh
github.com/mitchellh/reflectwalk v1.0.0/go.mod h1:mSTlrgnPZtwu0c4WaC2kGObEpuNDbx0jmZXqmk4esnw=
github.com/mmcloughlin/avo v0.0.0-20200803215136-443f81d77104 h1:ULR/QWMgcgRiZLUjSSJMU+fW+RDMstRdmnDWj9Q+AsA=
github.com/mmcloughlin/avo v0.0.0-20200803215136-443f81d77104/go.mod h1:wqKykBG2QzQDJEzvRkcS8x6MiSJkF52hXZsXcjaB3ls=
github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421 h1:ZqeYNhU3OHLH3mGKHDcjJRFFRrJa6eAM5H+CtDdOsPc=
github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg=
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742 h1:Esafd1046DLDQ0W1YjYsBW+p8U2u7vzgW2SQVmlNazg=
github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
github.com/modern-go/reflect2 v1.0.1 h1:9f412s+6RmYXLWZSEzVVgPGK7C2PphHj5RJrvfx9AWI=
github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
@ -349,6 +349,7 @@ github.com/nats-io/nats-streaming-server v0.18.0 h1:+RDozeN9scwCm0Wc2fYlvGcP144h
github.com/nats-io/nats-streaming-server v0.18.0/go.mod h1:Y9Aiif2oANuoKazQrs4wXtF3jqt6p97ODQg68lR5TnY=
github.com/nats-io/nats.go v1.10.0 h1:L8qnKaofSfNFbXg0C5F71LdjPRnmQwSsA4ukmkt1TvY=
github.com/nats-io/nats.go v1.10.0/go.mod h1:AjGArbfyR50+afOUotNX2Xs5SYHf+CoOa5HH1eEl2HE=
github.com/nats-io/nkeys v0.1.3 h1:6JrEfig+HzTH85yxzhSVbjHRJv9cn0p6n3IngIcM5/k=
github.com/nats-io/nkeys v0.1.3/go.mod h1:xpnFELMwJABBLVhffcfd1MZx6VsNRFpEugbxziKVo7w=
github.com/nats-io/nkeys v0.1.4 h1:aEsHIssIk6ETN5m2/MD8Y4B2X7FfXrBAUdkyRvbVYzA=
github.com/nats-io/nkeys v0.1.4/go.mod h1:XdZpAbhgyyODYqjTawOnIOI7VlbKSarI9Gfy1tqEu/s=
@ -367,6 +368,8 @@ github.com/opentracing/opentracing-go v1.1.0/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFSt
github.com/pascaldekloe/goe v0.1.0 h1:cBOtyMzM9HTpWjXfbbunk26uA6nG3a8n06Wieeh0MwY=
github.com/pascaldekloe/goe v0.1.0/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
github.com/pborman/getopt v0.0.0-20180729010549-6fdd0a2c7117/go.mod h1:85jBQOZwpVEaDAr341tbn15RS4fCAsIst0qp7i8ex1o=
github.com/philhofer/fwd v1.0.0 h1:UbZqGr5Y38ApvM/V/jEljVxwocdweyH+vmYvRPBnbqQ=
github.com/philhofer/fwd v1.0.0/go.mod h1:gk3iGcWd9+svBvR0sR+KPcfE+RNWozjowpeBVG3ZVNU=
github.com/philhofer/fwd v1.1.0 h1:PAdZw9+/BCf4gc/kA2L/PbGPkFe72Kl2GLZXTG8HpU8=
github.com/philhofer/fwd v1.1.0/go.mod h1:gk3iGcWd9+svBvR0sR+KPcfE+RNWozjowpeBVG3ZVNU=
github.com/pierrec/lz4 v2.0.5+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
@ -374,6 +377,7 @@ github.com/pierrec/lz4 v2.2.6+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi
github.com/pierrec/lz4 v2.4.0+incompatible h1:06usnXXDNcPvCHDkmPpkidf4jTc52UKld7UPfqKatY4=
github.com/pierrec/lz4 v2.4.0+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
@ -385,6 +389,7 @@ github.com/prometheus/client_golang v0.9.2/go.mod h1:OsXs2jCmiKlQ1lTBmv21f2mNfw4
github.com/prometheus/client_golang v1.0.0 h1:vrDKnkGzuGvhNAL56c7DBz29ZL+KxnoR0x7enabFceM=
github.com/prometheus/client_golang v1.0.0/go.mod h1:db9x61etRT2tGnBNRi70OPL5FsnadC4Ky3P0J6CfImo=
github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
github.com/prometheus/client_model v0.0.0-20190129233127-fd36f4220a90 h1:S/YWwWx/RA8rT8tKFRuGUZhuA90OyIBpPCXkcbwU8DE=
github.com/prometheus/client_model v0.0.0-20190129233127-fd36f4220a90/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4 h1:gQz4mCbXsO+nc9n1hCxHcGA3Zx3Eo+UHZoInFGUIXNM=
github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
@ -393,6 +398,7 @@ github.com/prometheus/common v0.4.1 h1:K0MGApIoQvMw27RTdJkPbr3JZ7DNbtxQNyi5STVM6
github.com/prometheus/common v0.4.1/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y86RQel1bk4=
github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
github.com/prometheus/procfs v0.0.0-20181204211112-1dc9a6cbc91a/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
github.com/prometheus/procfs v0.0.2 h1:6LJUbpNm42llc4HRCuvApCSWB/WfhuNo9K98Q9sNGfs=
github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA=
github.com/prometheus/procfs v0.0.8 h1:+fpWZdT24pJBiqJdAwYBjPSk+5YmQzYNPYzQsdzLkt8=
github.com/prometheus/procfs v0.0.8/go.mod h1:7Qr8sr6344vo1JqZ6HhLceV9o3AJ1Ff+GxbHq6oeK9A=
@ -417,6 +423,7 @@ github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPx
github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
github.com/sirupsen/logrus v1.6.0 h1:UBcNElsrwanuuMsnGSlYmtmgbb23qDR5dG+6X6Oo89I=
github.com/sirupsen/logrus v1.6.0/go.mod h1:7uNnSEd1DgxDLC74fIahvMZmmYsHGZGEOFrfsX/uA88=
github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d h1:zE9ykElWQ6/NYmHa3jpm/yHnI4xSofP+UP6SpjHcSeM=
github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc=
github.com/smartystreets/assertions v1.0.1 h1:voD4ITNjPL5jjBfgR/r8fPIIBrliWrWHeiJApdr3r4w=
github.com/smartystreets/assertions v1.0.1/go.mod h1:kHHU4qYBaI3q23Pp3VPrmWhuIUrLW/7eUrw0BU5VaoM=
@ -451,6 +458,8 @@ github.com/tinylib/msgp v1.1.3 h1:3giwAkmtaEDLSV0MdO1lDLuPgklgPzmk8H9+So2BVfA=
github.com/tinylib/msgp v1.1.3/go.mod h1:+d+yLhGm8mzTaHzB+wgMYrodPfmZrzkirds8fDWklFE=
github.com/tmc/grpc-websocket-proxy v0.0.0-20170815181823-89b8d40f7ca8 h1:ndzgwNDnKIqyCvHTXaCqh9KlOWKvBry6nuXMJmonVsE=
github.com/tmc/grpc-websocket-proxy v0.0.0-20170815181823-89b8d40f7ca8/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=
github.com/ttacon/chalk v0.0.0-20160626202418-22c06c80ed31 h1:OXcKh35JaYsGMRzpvFkLv/MEyPuL49CThT1pZ8aSml4=
github.com/ttacon/chalk v0.0.0-20160626202418-22c06c80ed31/go.mod h1:onvgF043R+lC5RZ8IT9rBXDaEDnpnw/Cl+HFiw+v/7Q=
github.com/tv42/httpunix v0.0.0-20150427012821-b75d8614f926/go.mod h1:9ESjWnEqriFuLhtthL60Sar/7RFoluCcXsuvEwTV5KM=
github.com/urfave/cli v1.20.0/go.mod h1:70zkFmudgCuE/ngEzBv17Jvp/497gISqfk5gWijbERA=
github.com/valyala/tcplisten v0.0.0-20161114210144-ceec8f93295a h1:0R4NLDRDZX6JcmhJgXi5E4b8Wg84ihbmUKp/GvSPEzc=
@ -471,6 +480,7 @@ 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=
go.etcd.io/etcd/v3 v3.3.0-rc.0.0.20200707003333-58bb8ae09f8e/go.mod h1:UENlOa05tkNvLx9VnNziSerG4Ro74upGK6Apd4v6M/Y=
go.opencensus.io v0.21.0 h1:mU6zScU4U1YAFPHEHYk+3JC4SY7JxgkqS10ZOSyksNg=
go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU=
go.opencensus.io v0.22.3 h1:8sGtKOrtQqkN1bp2AtX+misvLIlOmsEsNd+9NIcPEm8=
go.opencensus.io v0.22.3/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
@ -494,7 +504,9 @@ golang.org/x/crypto v0.0.0-20191002192127-34f69633bfdc/go.mod h1:yigFU9vqHzYiE8U
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
golang.org/x/crypto v0.0.0-20191206172530-e9b2fee46413/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
golang.org/x/crypto v0.0.0-20200323165209-0ec3e9974c59/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9 h1:psW17arqaxU48Z5kZ0CQnkZWQJsqcURM6tKiBApRjXI=
golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
golang.org/x/crypto v0.0.0-20200709230013-948cd5f35899 h1:DZhuSZLsGlFL4CmhA8BcRA0mnthyA/nZ00AqCUo7vHg=
golang.org/x/crypto v0.0.0-20200709230013-948cd5f35899/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
golang.org/x/crypto v0.0.0-20200820211705-5c72a883971a h1:vclmkQCjlDX5OydZ9wv8rBCcS0QyQY66Mpf/7BZbInM=
golang.org/x/crypto v0.0.0-20200820211705-5c72a883971a/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
@ -506,6 +518,7 @@ golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHl
golang.org/x/lint v0.0.0-20190930215403-16217165b5de h1:5hukYrvBGR8/eNkX5mdUezrA6JiaEZDtJb9Ei+1LlBs=
golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
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=
@ -523,6 +536,7 @@ golang.org/x/net v0.0.0-20190813141303-74dc4d7220e7/go.mod h1:z5CRVTTTmAJ677TzLL
golang.org/x/net v0.0.0-20191112182307-2180aed22343/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20200202094626-16171245cfb2/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-20200707034311-ab3426394381 h1:VXak5I6aEWmAXeQjA+QSZzlgNrpq9mjcfDemuexIKsU=
golang.org/x/net v0.0.0-20200707034311-ab3426394381/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
golang.org/x/net v0.0.0-20200822124328-c89045814202/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
golang.org/x/net v0.0.0-20200904194848-62affa334b73 h1:MXfv8rhZWmFeqX3GNZRsd6vOLoaCHjYEX3qkRo3YBUA=
@ -535,6 +549,7 @@ golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJ
golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20190227155943-e225da77a7e6/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
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=
@ -556,12 +571,15 @@ golang.org/x/sys v0.0.0-20190523142557-0e01d883c5c5/go.mod h1:h1NjWce9XRLGQEsW7w
golang.org/x/sys v0.0.0-20190726091711-fc99dfbffb4e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20191112214154-59a1497f0cea/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20200202164722-d101bd2416d5/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd h1:xhmwyvizuTgC2qz7ZlMluP20uW+C3Rm0FD/WLDX8884=
golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20200625212154-ddb9806d33ae h1:Ih9Yo4hSPImZOpfGuA4bR/ORKTAbhZo2AbWNRCnevdo=
golang.org/x/sys v0.0.0-20200625212154-ddb9806d33ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20200915084602-288bc346aa39 h1:356XA7ITklAU2//sYkjFeco+dH1bCRD8XCJ9FIEsvo4=
golang.org/x/sys v0.0.0-20200915084602-288bc346aa39/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.1-0.20181227161524-e6919f6577db h1:6/JqlYfC1CCaLnGceQTI+sDGhC9UBSPAsBqI0Gun6kU=
golang.org/x/text v0.3.1-0.20181227161524-e6919f6577db/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
golang.org/x/text v0.3.3 h1:cokOdA+Jmi5PJGXLlLllQSgYigAEfHXJAERHVMaCc2k=
@ -581,13 +599,16 @@ golang.org/x/tools v0.0.0-20190328211700-ab21143f2384/go.mod h1:LCzVGOaR6xXOjkQ3
golang.org/x/tools v0.0.0-20190524140312-2c0ae7006135/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
golang.org/x/tools v0.0.0-20190621195816-6e04913cbbac/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
golang.org/x/tools v0.0.0-20191029041327-9cc4af7d6b2c/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5 h1:hKsoRgsbwY1NafxrwTs+k64bikrLBkAgPir1TNCj3Zs=
golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
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-20200929223013-bf155c11ec6f h1:7+Nz9MyPqt2qMCTvNiRy1G0zYfkB7UCa+ayT6uVvbyI=
golang.org/x/tools v0.0.0-20200929223013-bf155c11ec6f/go.mod h1:z6u4i615ZeAfBE4XtMziQW1fSVJXACjjbWkB/mvPzlU=
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=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 h1:go1bK/D/BFZV2I8cIQd1NKEZ+0owSTG1fDTci4IqFcE=
golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
@ -600,12 +621,14 @@ google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoA
google.golang.org/genproto v0.0.0-20190307195333-5fe7a883aa19/go.mod h1:VzzqZJRnGkLBvHegQrXjBqPurQTc5/KpmUdxsrq26oE=
google.golang.org/genproto v0.0.0-20190404172233-64821d5d2107/go.mod h1:VzzqZJRnGkLBvHegQrXjBqPurQTc5/KpmUdxsrq26oE=
google.golang.org/genproto v0.0.0-20190425155659-357c62f0e4bb/go.mod h1:VzzqZJRnGkLBvHegQrXjBqPurQTc5/KpmUdxsrq26oE=
google.golang.org/genproto v0.0.0-20190508193815-b515fa19cec8 h1:x913Lq/RebkvUmRSdQ8MNb0GZKn+SR1ESfoetcQSeak=
google.golang.org/genproto v0.0.0-20190508193815-b515fa19cec8/go.mod h1:VzzqZJRnGkLBvHegQrXjBqPurQTc5/KpmUdxsrq26oE=
google.golang.org/genproto v0.0.0-20190819201941-24fa4b261c55 h1:gSJIx1SDwno+2ElGhA4+qG2zF97qiUzTM+rQ0klBOcE=
google.golang.org/genproto v0.0.0-20190819201941-24fa4b261c55/go.mod h1:DMBHOl98Agz4BDEuKkezgsaosCRResVns1a3J2ZsMNc=
google.golang.org/grpc v1.14.0/go.mod h1:yo6s7OP7yaDglbqo1J04qKzAhqBH6lvTonzMVmEdcZw=
google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
google.golang.org/grpc v1.20.1/go.mod h1:10oTOabMzJvdu6/UiuZezV6QK5dSlG84ov/aaiqXj38=
google.golang.org/grpc v1.22.0 h1:J0UbZOIrCAl+fpTOf8YLs4dJo8L/owV4LYVtAXQoPkw=
google.golang.org/grpc v1.22.0/go.mod h1:Y5yQAOtifL1yxbo5wqy6BxZv8vAUGQwXBOALyacEbxg=
google.golang.org/grpc v1.23.0/go.mod h1:Y5yQAOtifL1yxbo5wqy6BxZv8vAUGQwXBOALyacEbxg=
google.golang.org/grpc v1.26.0 h1:2dTRdpdFEEhJYQD8EMLB61nnrzSCTbG38PhqdhvOltg=
@ -634,6 +657,7 @@ gopkg.in/jcmturner/dnsutils.v1 v1.0.1 h1:cIuC1OLRGZrld+16ZJvvZxVJeKPsvd5eUIvxfoN
gopkg.in/jcmturner/dnsutils.v1 v1.0.1/go.mod h1:m3v+5svpVOhtFAP/wSz+yzh4Mc0Fg7eRhxkJMWSIz9Q=
gopkg.in/jcmturner/goidentity.v3 v3.0.0 h1:1duIyWiTaYvVx3YX2CYtpJbUFd7/UuPYCfgXtQ3VTbI=
gopkg.in/jcmturner/goidentity.v3 v3.0.0/go.mod h1:oG2kH0IvSYNIu80dVAyu/yoefjq1mNfM5bm88whjWx4=
gopkg.in/jcmturner/gokrb5.v7 v7.2.3 h1:hHMV/yKPwMnJhPuPx7pH2Uw/3Qyf+thJYlisUc44010=
gopkg.in/jcmturner/gokrb5.v7 v7.2.3/go.mod h1:l8VISx+WGYp+Fp7KRbsiUuXTTOnxIc3Tuvyavf11/WM=
gopkg.in/jcmturner/gokrb5.v7 v7.3.0 h1:0709Jtq/6QXEuWRfAm260XqlpcwL1vxtO1tUE2qK8Z4=
gopkg.in/jcmturner/gokrb5.v7 v7.3.0/go.mod h1:l8VISx+WGYp+Fp7KRbsiUuXTTOnxIc3Tuvyavf11/WM=

15
pkg/bucket/replication/replication.go
Unescape View File

@ -46,11 +46,6 @@ func (s StatusType) String() string {
return string(s)
}
// Empty returns true if this status is not set
func (s StatusType) Empty() bool {
return string(s) == ""
}
var (
errReplicationTooManyRules = Errorf("Replication configuration allows a maximum of 1000 rules")
errReplicationNoRule = Errorf("Replication configuration should have at least one rule")
@ -161,11 +156,8 @@ func (c Config) GetDestination() Destination {
func (c Config) Replicate(obj ObjectOpts) bool {
for _, rule := range c.FilterActionableRules(obj) {
// check MinIO extension for versioned deletes
if !obj.DeleteMarker && obj.VersionID != "" && rule.DeleteReplication.Status == Disabled {
return false
}
if obj.DeleteMarker && rule.DeleteMarkerReplication.Status == Disabled {
if obj.DeleteMarker {
// Indicates whether MinIO will remove a delete marker. By default, delete markers
// are not replicated.
return false
@ -173,6 +165,9 @@ func (c Config) Replicate(obj ObjectOpts) bool {
if obj.SSEC {
return false
}
if obj.VersionID != "" && !obj.IsLatest {
return false
}
if rule.Status == Disabled {
continue
}

51
pkg/bucket/replication/rule.go
Unescape View File

@ -45,50 +45,12 @@ func (d DeleteMarkerReplication) Validate() error {
if d.IsEmpty() {
return errDeleteMarkerReplicationMissing
}
if d.Status != Disabled && d.Status != Enabled {
if d.Status != Disabled {
return errInvalidDeleteMarkerReplicationStatus
}
return nil
}
// DeleteReplication - whether versioned deletes are replicated - this is a MinIO only
// extension.
type DeleteReplication struct {
Status Status `xml:"Status"` // should be set to "Disabled" by default
}
// IsEmpty returns true if DeleteReplication is not set
func (d DeleteReplication) IsEmpty() bool {
return len(d.Status) == 0
}
// Validate validates whether the status is disabled.
func (d DeleteReplication) Validate() error {
if d.IsEmpty() {
return errDeleteReplicationMissing
}
if d.Status != Disabled && d.Status != Enabled {
return errInvalidDeleteReplicationStatus
}
return nil
}
// UnmarshalXML - decodes XML data.
func (d *DeleteReplication) UnmarshalXML(dec *xml.Decoder, start xml.StartElement) (err error) {
// Make subtype to avoid recursive UnmarshalXML().
type deleteReplication DeleteReplication
drep := deleteReplication{}
if err := dec.DecodeElement(&drep, &start); err != nil {
return err
}
if len(drep.Status) == 0 {
drep.Status = Disabled
}
d.Status = drep.Status
return nil
}
// Rule - a rule for replication configuration.
type Rule struct {
XMLName xml.Name `xml:"Rule" json:"Rule"`
@ -96,10 +58,8 @@ type Rule struct {
Status Status `xml:"Status" json:"Status"`
Priority int `xml:"Priority" json:"Priority"`
DeleteMarkerReplication DeleteMarkerReplication `xml:"DeleteMarkerReplication" json:"DeleteMarkerReplication"`
// MinIO extension to replicate versioned deletes
DeleteReplication DeleteReplication `xml:"DeleteReplication" json:"DeleteReplication"`
Destination Destination `xml:"Destination" json:"Destination"`
Filter Filter `xml:"Filter" json:"Filter"`
Destination Destination `xml:"Destination" json:"Destination"`
Filter Filter `xml:"Filter" json:"Filter"`
}
var (
@ -110,8 +70,6 @@ var (
errPriorityMissing = Errorf("Priority must be specified")
errInvalidDeleteMarkerReplicationStatus = Errorf("Delete marker replication is currently not supported")
errDestinationSourceIdentical = Errorf("Destination bucket cannot be the same as the source bucket.")
errDeleteReplicationMissing = Errorf("Delete replication must be specified")
errInvalidDeleteReplicationStatus = Errorf("Delete replication is either enable|disable")
)
// validateID - checks if ID is valid or not.
@ -188,9 +146,6 @@ func (r Rule) Validate(bucket string, sameTarget bool) error {
if err := r.DeleteMarkerReplication.Validate(); err != nil {
return err
}
if err := r.DeleteReplication.Validate(); err != nil {
return err
}
if r.Priority < 0 {
return errPriorityMissing
}

Write Preview
Loading…
Cancel
Save