fix: avoid sending errors on missing objects on locked buckets (#10994)

make sure multi-object delete returned errors that are AWS S3 compatible
master
Harshavardhana 4 years ago committed by GitHub
parent e6fa410778
commit bdd094bc39
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 16
      cmd/api-datatypes.go
  2. 10
      cmd/api-errors.go
  3. 3
      cmd/bucket-handlers.go
  4. 5
      cmd/bucket-object-lock.go
  5. 2
      cmd/bucket-replication.go
  6. 2
      cmd/data-crawler.go
  7. 65
      cmd/erasure-object.go
  8. 14
      cmd/erasure-server-sets.go
  9. 4
      cmd/erasure-sets.go
  10. 9
      cmd/object-api-errors.go
  11. 4
      cmd/object-api-options.go
  12. 4
      cmd/object-handlers.go
  13. 4
      cmd/object-handlers_test.go
  14. 2
      cmd/web-handlers.go
  15. 13
      cmd/xl-storage-format-v2.go
  16. 18
      cmd/xl-storage.go

@ -32,13 +32,27 @@ type DeletedObject struct {
// Replication status of DeleteMarker
DeleteMarkerReplicationStatus string `xml:"DeleteMarkerReplicationStatus,omitempty"`
// MTime of DeleteMarker on source that needs to be propagated to replica
DeleteMarkerMTime time.Time `xml:"DeleteMarkerMTime,omitempty"`
DeleteMarkerMTime DeleteMarkerMTime `xml:"DeleteMarkerMTime,omitempty"`
// Status of versioned delete (of object or DeleteMarker)
VersionPurgeStatus VersionPurgeStatusType `xml:"VersionPurgeStatus,omitempty"`
// PurgeTransitioned is nonempty if object is in transition tier
PurgeTransitioned string `xml:"PurgeTransitioned,omitempty"`
}
// DeleteMarkerMTime is an embedded type containing time.Time for XML marshal
type DeleteMarkerMTime struct {
time.Time
}
// MarshalXML encodes expiration date if it is non-zero and encodes
// empty string otherwise
func (t DeleteMarkerMTime) MarshalXML(e *xml.Encoder, startElement xml.StartElement) error {
if t.Time.IsZero() {
return nil
}
return e.EncodeElement(t.Time.Format(time.RFC3339), startElement)
}
// ObjectToDelete carries key name for the object to delete.
type ObjectToDelete struct {
ObjectName string `xml:"Key"`

@ -124,6 +124,7 @@ const (
ErrObjectRestoreAlreadyInProgress
ErrNoSuchKey
ErrNoSuchUpload
ErrInvalidVersionID
ErrNoSuchVersion
ErrNotImplemented
ErrPreconditionFailed
@ -558,11 +559,16 @@ var errorCodes = errorCodeMap{
Description: "The specified multipart upload does not exist. The upload ID may be invalid, or the upload may have been aborted or completed.",
HTTPStatusCode: http.StatusNotFound,
},
ErrNoSuchVersion: {
ErrInvalidVersionID: {
Code: "InvalidArgument",
Description: "Invalid version id specified",
HTTPStatusCode: http.StatusBadRequest,
},
ErrNoSuchVersion: {
Code: "NoSuchVersion",
Description: "The specified version does not exist.",
HTTPStatusCode: http.StatusNotFound,
},
ErrNotImplemented: {
Code: "NotImplemented",
Description: "A header you provided implies functionality that is not implemented",
@ -1886,6 +1892,8 @@ func toAPIErrorCode(ctx context.Context, err error) (apiErr APIErrorCode) {
apiErr = ErrNoSuchKey
case MethodNotAllowed:
apiErr = ErrMethodNotAllowed
case InvalidVersionID:
apiErr = ErrInvalidVersionID
case VersionNotFound:
apiErr = ErrNoSuchVersion
case ObjectAlreadyExists:

@ -495,7 +495,6 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
})
deletedObjects := make([]DeletedObject, len(deleteObjects.Objects))
for i := range errs {
dindex := objectsToDelete[ObjectToDelete{
ObjectName: dObjects[i].ObjectName,
VersionID: dObjects[i].VersionID,
@ -504,7 +503,7 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
DeleteMarkerReplicationStatus: dObjects[i].DeleteMarkerReplicationStatus,
PurgeTransitioned: dObjects[i].PurgeTransitioned,
}]
if errs[i] == nil || isErrObjectNotFound(errs[i]) || isErrVersionNotFound(errs[i]) {
if errs[i] == nil || isErrObjectNotFound(errs[i]) {
if replicateDeletes {
dObjects[i].DeleteMarkerReplicationStatus = deleteList[i].DeleteMarkerReplicationStatus
dObjects[i].VersionPurgeStatus = deleteList[i].VersionPurgeStatus

@ -89,16 +89,15 @@ func enforceRetentionBypassForDelete(ctx context.Context, r *http.Request, bucke
}
opts.VersionID = object.VersionID
if gerr != nil { // error from GetObjectInfo
switch err.(type) {
switch gerr.(type) {
case MethodNotAllowed: // This happens usually for a delete marker
if oi.DeleteMarker {
// Delete marker should be present and valid.
return ErrNone
}
}
return toAPIErrorCode(ctx, err)
return toAPIErrorCode(ctx, gerr)
}
lhold := objectlock.GetObjectLegalHoldMeta(oi.UserDefined)

@ -200,7 +200,7 @@ func replicateDelete(ctx context.Context, dobj DeletedObjectVersionInfo, objectA
VersionID: versionID,
Internal: miniogo.AdvancedRemoveOptions{
ReplicationDeleteMarker: dobj.DeleteMarkerVersionID != "",
ReplicationMTime: dobj.DeleteMarkerMTime,
ReplicationMTime: dobj.DeleteMarkerMTime.Time,
ReplicationStatus: miniogo.ReplicationStatusReplica,
},
})

@ -854,7 +854,7 @@ func (i *crawlItem) healReplicationDeletes(ctx context.Context, o ObjectLayer, m
DeleteMarkerVersionID: dmVersionID,
VersionID: versionID,
DeleteMarkerReplicationStatus: string(meta.oi.ReplicationStatus),
DeleteMarkerMTime: meta.oi.ModTime,
DeleteMarkerMTime: DeleteMarkerMTime{meta.oi.ModTime},
DeleteMarker: meta.oi.DeleteMarker,
VersionPurgeStatus: meta.oi.VersionPurgeStatus,
},

@ -18,7 +18,6 @@ package cmd
import (
"context"
"errors"
"fmt"
"io"
"net/http"
@ -857,19 +856,21 @@ func (er erasureObjects) DeleteObjects(ctx context.Context, bucket string, objec
}
// Initialize list of errors.
var opErrs = make([]error, len(storageDisks))
var delObjErrs = make([][]error, len(storageDisks))
var wg sync.WaitGroup
// Remove versions in bulk for each disk
for index, disk := range storageDisks {
if disk == nil {
opErrs[index] = errDiskNotFound
continue
}
wg.Add(1)
go func(index int, disk StorageAPI) {
defer wg.Done()
if disk == nil {
delObjErrs[index] = make([]error, len(versions))
for i := range versions {
delObjErrs[index][i] = errDiskNotFound
}
return
}
delObjErrs[index] = disk.DeleteVersions(ctx, bucket, versions)
}(index, disk)
}
@ -878,43 +879,43 @@ func (er erasureObjects) DeleteObjects(ctx context.Context, bucket string, objec
// Reduce errors for each object
for objIndex := range objects {
if errs[objIndex] != nil {
continue
}
diskErrs := make([]error, len(storageDisks))
// Iterate over disks to fetch the error
// of deleting of the current object
for i := range delObjErrs {
// delObjErrs[i] is not nil when disks[i] is also not nil
if delObjErrs[i] != nil {
if errors.Is(delObjErrs[i][objIndex], errFileNotFound) ||
errors.Is(delObjErrs[i][objIndex], errFileVersionNotFound) {
continue
}
diskErrs[i] = delObjErrs[i][objIndex]
}
}
errs[objIndex] = reduceWriteQuorumErrs(ctx, diskErrs, objectOpIgnoredErrs, writeQuorums[objIndex])
err := reduceWriteQuorumErrs(ctx, diskErrs, objectOpIgnoredErrs, writeQuorums[objIndex])
if objects[objIndex].VersionID != "" {
errs[objIndex] = toObjectErr(err, bucket, objects[objIndex].ObjectName, objects[objIndex].VersionID)
} else {
errs[objIndex] = toObjectErr(err, bucket, objects[objIndex].ObjectName)
}
if errs[objIndex] == nil {
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,
PurgeTransitioned: objects[objIndex].PurgeTransitioned,
}
} else {
dobjects[objIndex] = DeletedObject{
ObjectName: versions[objIndex].Name,
VersionID: versions[objIndex].VersionID,
VersionPurgeStatus: versions[objIndex].VersionPurgeStatus,
DeleteMarkerReplicationStatus: versions[objIndex].DeleteMarkerReplicationStatus,
PurgeTransitioned: objects[objIndex].PurgeTransitioned,
}
}
if versions[objIndex].Deleted {
dobjects[objIndex] = DeletedObject{
DeleteMarker: versions[objIndex].Deleted,
DeleteMarkerVersionID: versions[objIndex].VersionID,
DeleteMarkerMTime: DeleteMarkerMTime{versions[objIndex].ModTime},
DeleteMarkerReplicationStatus: versions[objIndex].DeleteMarkerReplicationStatus,
ObjectName: versions[objIndex].Name,
VersionPurgeStatus: versions[objIndex].VersionPurgeStatus,
PurgeTransitioned: objects[objIndex].PurgeTransitioned,
}
} else {
dobjects[objIndex] = DeletedObject{
ObjectName: versions[objIndex].Name,
VersionID: versions[objIndex].VersionID,
VersionPurgeStatus: versions[objIndex].VersionPurgeStatus,
DeleteMarkerReplicationStatus: versions[objIndex].DeleteMarkerReplicationStatus,
PurgeTransitioned: objects[objIndex].PurgeTransitioned,
}
}
}

@ -585,17 +585,17 @@ func (z *erasureServerSets) DeleteObjects(ctx context.Context, bucket string, ob
}
defer multiDeleteLock.Unlock()
if z.SingleZone() {
return z.serverSets[0].DeleteObjects(ctx, bucket, objects, opts)
}
for _, zone := range z.serverSets {
deletedObjects, errs := zone.DeleteObjects(ctx, bucket, objects, opts)
for i, derr := range errs {
if derrs[i] == nil {
if derr != nil && !isErrObjectNotFound(derr) && !isErrVersionNotFound(derr) {
derrs[i] = derr
}
}
if derrs[i] == nil {
dobjects[i] = deletedObjects[i]
if derr != nil {
derrs[i] = derr
}
dobjects[i] = deletedObjects[i]
}
}
return dobjects, derrs

@ -759,9 +759,7 @@ func (s *erasureSets) DeleteObjects(ctx context.Context, bucket string, objects
dobjects, errs := s.getHashedSet(objsGroup[0].object.ObjectName).DeleteObjects(ctx, bucket, toNames(objsGroup), opts)
for i, obj := range objsGroup {
delErrs[obj.origIndex] = errs[i]
if delErrs[obj.origIndex] == nil {
delObjects[obj.origIndex] = dobjects[i]
}
delObjects[obj.origIndex] = dobjects[i]
}
}

@ -255,7 +255,14 @@ func (e BucketNotEmpty) Error() string {
return "Bucket not empty: " + e.Bucket
}
// VersionNotFound object does not exist.
// InvalidVersionID invalid version id
type InvalidVersionID GenericError
func (e InvalidVersionID) Error() string {
return "Invalid version id: " + e.Bucket + "/" + e.Object + "(" + e.VersionID + ")"
}
// VersionNotFound version does not exist.
type VersionNotFound GenericError
func (e VersionNotFound) Error() string {

@ -93,7 +93,7 @@ func getOpts(ctx context.Context, r *http.Request, bucket, object string) (Objec
_, err := uuid.Parse(vid)
if err != nil {
logger.LogIf(ctx, err)
return opts, VersionNotFound{
return opts, InvalidVersionID{
Bucket: bucket,
Object: object,
VersionID: vid,
@ -209,7 +209,7 @@ func putOpts(ctx context.Context, r *http.Request, bucket, object string, metada
_, err := uuid.Parse(vid)
if err != nil {
logger.LogIf(ctx, err)
return opts, VersionNotFound{
return opts, InvalidVersionID{
Bucket: bucket,
Object: object,
VersionID: vid,

@ -2798,13 +2798,14 @@ func (api objectAPIHandlers) DeleteObjectHandler(w http.ResponseWriter, r *http.
VersionID: versionID,
DeleteMarkerVersionID: dmVersionID,
DeleteMarkerReplicationStatus: string(objInfo.ReplicationStatus),
DeleteMarkerMTime: objInfo.ModTime,
DeleteMarkerMTime: DeleteMarkerMTime{objInfo.ModTime},
DeleteMarker: objInfo.DeleteMarker,
VersionPurgeStatus: objInfo.VersionPurgeStatus,
},
Bucket: bucket,
})
}
if goi.TransitionStatus == lifecycle.TransitionComplete { // clean up transitioned tier
action := lifecycle.DeleteAction
if goi.VersionID != "" {
@ -2819,6 +2820,7 @@ func (api objectAPIHandlers) DeleteObjectHandler(w http.ResponseWriter, r *http.
IsLatest: goi.IsLatest,
}, action, true)
}
setPutObjHeaders(w, objInfo, true)
writeSuccessNoContent(w)
}

@ -1765,7 +1765,7 @@ func testAPICopyObjectPartHandler(obj ObjectLayer, instanceType, bucketName stri
copySourceHeader: url.QueryEscape(SlashSeparator+bucketName+SlashSeparator+objectName) + "?versionId=17",
accessKey: credentials.AccessKey,
secretKey: credentials.SecretKey,
expectedRespStatus: http.StatusBadRequest,
expectedRespStatus: http.StatusNotFound,
},
}
@ -2135,7 +2135,7 @@ func testAPICopyObjectHandler(obj ObjectLayer, instanceType, bucketName string,
copySourceHeader: url.QueryEscape(SlashSeparator+bucketName+SlashSeparator+objectName) + "?versionId=17",
accessKey: credentials.AccessKey,
secretKey: credentials.SecretKey,
expectedRespStatus: http.StatusBadRequest,
expectedRespStatus: http.StatusNotFound,
},
}

@ -761,7 +761,7 @@ next:
ObjectName: objectName,
DeleteMarkerVersionID: oi.VersionID,
DeleteMarkerReplicationStatus: string(oi.ReplicationStatus),
DeleteMarkerMTime: oi.ModTime,
DeleteMarkerMTime: DeleteMarkerMTime{oi.ModTime},
DeleteMarker: oi.DeleteMarker,
VersionPurgeStatus: oi.VersionPurgeStatus,
},

@ -445,8 +445,12 @@ func (z *xlMetaV2) DeleteVersion(fi FileInfo) (string, bool, error) {
}
var uv uuid.UUID
var err error
if fi.VersionID != "" {
uv, _ = uuid.Parse(fi.VersionID)
uv, err = uuid.Parse(fi.VersionID)
if err != nil {
return "", false, errFileVersionNotFound
}
}
var ventry xlMetaV2Version
@ -645,8 +649,11 @@ func (z xlMetaV2) ListVersions(volume, path string) (versions []FileInfo, modTim
// for consumption across callers.
func (z xlMetaV2) ToFileInfo(volume, path, versionID string) (fi FileInfo, err error) {
var uv uuid.UUID
if versionID != "" {
uv, _ = uuid.Parse(versionID)
if versionID != "" && versionID != nullVersionID {
uv, err = uuid.Parse(versionID)
if err != nil {
return FileInfo{}, errFileVersionNotFound
}
}
var latestModTime time.Time

@ -971,10 +971,16 @@ func (s *xlStorage) DeleteVersion(ctx context.Context, volume, path string, fi F
buf, err := s.ReadAll(ctx, volume, pathJoin(path, xlStorageFormatFile))
if err != nil {
if err == errFileNotFound && fi.VersionID != "" {
err = errFileVersionNotFound
}
return err
}
if len(buf) == 0 {
if fi.VersionID != "" {
return errFileVersionNotFound
}
return errFileNotFound
}
@ -1146,10 +1152,22 @@ func (s *xlStorage) ReadVersion(ctx context.Context, volume, path, versionID str
if err != nil {
if err == errFileNotFound {
if err = s.renameLegacyMetadata(volume, path); err != nil {
if err == errFileNotFound {
if versionID != "" {
return fi, errFileVersionNotFound
}
return fi, errFileNotFound
}
return fi, err
}
buf, err = s.ReadAll(ctx, volume, pathJoin(path, xlStorageFormatFile))
if err != nil {
if err == errFileNotFound {
if versionID != "" {
return fi, errFileVersionNotFound
}
return fi, errFileNotFound
}
return fi, err
}
} else {

Loading…
Cancel
Save