avoid notification for non-existent delete objects (#11514)

Skip notifications on objects that might have had
an error during deletion, this also avoids unnecessary
replication attempt on such objects.

Refactor some places to make sure that we have notified
the client before we

- notify
- schedule for replication
- lifecycle etc.

cmd/bucket-handlers.go

@@ -583,6 +583,10 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
 	// Write success response.
 	writeSuccessResponseXML(w, encodedSuccessResponse)
 	for _, dobj := range deletedObjects {
+		if dobj.ObjectName == "" {
+			continue
+		}
+
 		if replicateDeletes {
 			if dobj.DeleteMarkerReplicationStatus == string(replication.Pending) || dobj.VersionPurgeStatus == Pending {
 				dv := DeletedObjectVersionInfo{
@@ -594,18 +598,14 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
 		}
 
 		if hasLifecycleConfig && dobj.PurgeTransitioned == lifecycle.TransitionComplete { // clean up transitioned tier
-			deleteTransitionedObject(ctx, newObjectLayerFn(), bucket, dobj.ObjectName, lifecycle.ObjectOpts{
+			deleteTransitionedObject(ctx, objectAPI, bucket, dobj.ObjectName, lifecycle.ObjectOpts{
 				Name:         dobj.ObjectName,
 				VersionID:    dobj.VersionID,
 				DeleteMarker: dobj.DeleteMarker,
 			}, false, true)
 		}
 
-	}
-	// Notify deleted event for objects.
-	for _, dobj := range deletedObjects {
 		eventName := event.ObjectRemovedDelete
-
 		objInfo := ObjectInfo{
 			Name:      dobj.ObjectName,
 			VersionID: dobj.VersionID,

cmd/bucket-replication.go

@@ -283,7 +283,7 @@ func replicateDelete(ctx context.Context, dobj DeletedObjectVersionInfo, objectA
 		} else {
 			versionPurgeStatus = Failed
 		}
-		logger.LogIf(ctx, fmt.Errorf("Unable to replicate delete marker to %s/%s(%s): %s", rcfg.GetDestination().Bucket, dobj.ObjectName, versionID, err))
+		logger.LogIf(ctx, fmt.Errorf("Unable to replicate delete marker to %s/%s(%s): %s", rcfg.GetDestination().Bucket, dobj.ObjectName, versionID, rmErr))
 	} else {
 		if dobj.VersionID == "" {
 			replicationStatus = string(replication.Completed)

cmd/handler-api.go

@@ -60,7 +60,7 @@ func (t *apiConfig) init(cfg api.Config, setDriveCounts []int) {
 		}
 		// max requests per node is calculated as
 		// total_ram / ram_per_request
-		// ram_per_request is 1MiB * driveCount + 2 * 10MiB (default erasure block size)
+		// ram_per_request is (2MiB+128KiB) * driveCount + 2 * 10MiB (default erasure block size)
 		apiRequestsMaxPerNode = int(stats.TotalRAM / uint64(t.totalDriveCount*(blockSizeLarge+blockSizeSmall)+blockSizeV1*2))
 	} else {
 		apiRequestsMaxPerNode = cfg.RequestsMax

cmd/object-handlers-common.go

@@ -26,8 +26,6 @@ import (
 
 	xhttp "github.com/minio/minio/cmd/http"
 	"github.com/minio/minio/pkg/bucket/lifecycle"
-	"github.com/minio/minio/pkg/event"
-	"github.com/minio/minio/pkg/handlers"
 )
 
 var (
@@ -261,7 +259,7 @@ func setPutObjHeaders(w http.ResponseWriter, objInfo ObjectInfo, delete bool) {
 		}
 	}
 
-	if objInfo.Bucket != "" {
+	if objInfo.Bucket != "" && objInfo.Name != "" {
 		if lc, err := globalLifecycleSys.Get(objInfo.Bucket); err == nil && !delete {
 			ruleID, expiryTime := lc.PredictExpiryTime(lifecycle.ObjectOpts{
 				Name:         objInfo.Name,
@@ -279,32 +277,3 @@ func setPutObjHeaders(w http.ResponseWriter, objInfo ObjectInfo, delete bool) {
 		}
 	}
 }
-
-// deleteObject is a convenient wrapper to delete an object, this
-// is a common function to be called from object handlers and
-// web handlers.
-func deleteObject(ctx context.Context, obj ObjectLayer, cache CacheObjectLayer, bucket, object string, w http.ResponseWriter, r *http.Request, opts ObjectOptions) (objInfo ObjectInfo, err error) {
-	deleteObject := obj.DeleteObject
-	if cache != nil {
-		deleteObject = cache.DeleteObject
-	}
-	// Proceed to delete the object.
-	objInfo, err = deleteObject(ctx, bucket, object, opts)
-	if objInfo.Name != "" {
-		eventName := event.ObjectRemovedDelete
-		if objInfo.DeleteMarker {
-			eventName = event.ObjectRemovedDeleteMarkerCreated
-		}
-		// Notify object deleted marker event.
-		sendEvent(eventArgs{
-			EventName:    eventName,
-			BucketName:   bucket,
-			Object:       objInfo,
-			ReqParams:    extractReqParams(r),
-			RespElements: extractRespElements(w),
-			UserAgent:    r.UserAgent(),
-			Host:         handlers.GetSourceIP(r),
-		})
-	}
-	return objInfo, err
-}

cmd/object-handlers.go

@@ -2848,8 +2848,13 @@ func (api objectAPIHandlers) DeleteObjectHandler(w http.ResponseWriter, r *http.
 		return
 	}
 
+	deleteObject := objectAPI.DeleteObject
+	if api.CacheAPI() != nil {
+		deleteObject = api.CacheAPI().DeleteObject
+	}
+
 	// http://docs.aws.amazon.com/AmazonS3/latest/API/RESTObjectDELETE.html
-	objInfo, err := deleteObject(ctx, objectAPI, api.CacheAPI(), bucket, object, w, r, opts)
+	objInfo, err := deleteObject(ctx, bucket, object, opts)
 	if err != nil {
 		switch err.(type) {
 		case BucketNotFound:
@@ -2857,9 +2862,32 @@ func (api objectAPIHandlers) DeleteObjectHandler(w http.ResponseWriter, r *http.
 			writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
 			return
 		}
-		// Ignore delete object errors while replying to client, since we are suppposed to reply only 204.
 	}
 
+	if objInfo.Name == "" {
+		writeSuccessNoContent(w)
+		return
+	}
+
+	setPutObjHeaders(w, objInfo, true)
+	writeSuccessNoContent(w)
+
+	eventName := event.ObjectRemovedDelete
+	if objInfo.DeleteMarker {
+		eventName = event.ObjectRemovedDeleteMarkerCreated
+	}
+
+	// Notify object deleted event.
+	sendEvent(eventArgs{
+		EventName:    eventName,
+		BucketName:   bucket,
+		Object:       objInfo,
+		ReqParams:    extractReqParams(r),
+		RespElements: extractRespElements(w),
+		UserAgent:    r.UserAgent(),
+		Host:         handlers.GetSourceIP(r),
+	})
+
 	if replicateDel {
 		dmVersionID := ""
 		versionID := ""
@@ -2884,7 +2912,7 @@ func (api objectAPIHandlers) DeleteObjectHandler(w http.ResponseWriter, r *http.
 	}
 
 	if goi.TransitionStatus == lifecycle.TransitionComplete { // clean up transitioned tier
-		deleteTransitionedObject(ctx, newObjectLayerFn(), bucket, object, lifecycle.ObjectOpts{
+		deleteTransitionedObject(ctx, objectAPI, bucket, object, lifecycle.ObjectOpts{
 			Name:             object,
 			UserTags:         goi.UserTags,
 			VersionID:        goi.VersionID,
@@ -2893,9 +2921,6 @@ func (api objectAPIHandlers) DeleteObjectHandler(w http.ResponseWriter, r *http.
 			IsLatest:         goi.IsLatest,
 		}, false, true)
 	}
-
-	setPutObjHeaders(w, objInfo, true)
-	writeSuccessNoContent(w)
 }
 
 // PutObjectLegalHoldHandler - set legal hold configuration to object,

cmd/web-handlers.go

@@ -761,8 +761,39 @@ next:
 				}
 			}
 
-			oi, err := deleteObject(ctx, objectAPI, web.CacheAPI(), args.BucketName, objectName, nil, r, opts)
-			if replicateDel && err == nil {
+			deleteObject := objectAPI.DeleteObject
+			if web.CacheAPI() != nil {
+				deleteObject = web.CacheAPI().DeleteObject
+			}
+
+			oi, err := deleteObject(ctx, args.BucketName, objectName, opts)
+			if err != nil {
+				switch err.(type) {
+				case BucketNotFound:
+					return toJSONError(ctx, err)
+				}
+			}
+			if oi.Name == "" {
+				logger.LogIf(ctx, err)
+				continue
+			}
+
+			eventName := event.ObjectRemovedDelete
+			if oi.DeleteMarker {
+				eventName = event.ObjectRemovedDeleteMarkerCreated
+			}
+
+			// Notify object deleted event.
+			sendEvent(eventArgs{
+				EventName:  eventName,
+				BucketName: args.BucketName,
+				Object:     oi,
+				ReqParams:  extractReqParams(r),
+				UserAgent:  r.UserAgent(),
+				Host:       handlers.GetSourceIP(r),
+			})
+
+			if replicateDel {
 				dobj := DeletedObjectVersionInfo{
 					DeletedObject: DeletedObject{
 						ObjectName:                    objectName,
@@ -776,12 +807,13 @@ next:
 				}
 				scheduleReplicationDelete(ctx, dobj, objectAPI, replicateSync)
 			}
-			if goi.TransitionStatus == lifecycle.TransitionComplete && err == nil && goi.VersionID == "" {
-				deleteTransitionedObject(ctx, newObjectLayerFn(), args.BucketName, objectName, lifecycle.ObjectOpts{
+			if goi.TransitionStatus == lifecycle.TransitionComplete {
+				deleteTransitionedObject(ctx, objectAPI, args.BucketName, objectName, lifecycle.ObjectOpts{
 					Name:             objectName,
 					UserTags:         goi.UserTags,
 					VersionID:        goi.VersionID,
 					DeleteMarker:     goi.DeleteMarker,
+					TransitionStatus: goi.TransitionStatus,
 					IsLatest:         goi.IsLatest,
 				}, false, true)
 			}

cmd/xl-storage.go

@@ -53,7 +53,7 @@ import (
 
 const (
 	nullVersionID  = "null"
-	blockSizeLarge = 1 * humanize.MiByte   // Default r/w block size for larger objects.
+	blockSizeLarge = 2 * humanize.MiByte   // Default r/w block size for larger objects.
 	blockSizeSmall = 128 * humanize.KiByte // Default r/w block size for smaller objects.
 
 	// On regular files bigger than this;