From ee158e16108796d1c1863f869d5d7ce9511850bd Mon Sep 17 00:00:00 2001
From: Krishnan Parthasarathi <krisis@users.noreply.github.com>
Date: Thu, 22 Feb 2024 15:00:32 -0800
Subject: [PATCH] ilm: Update action count only on success (#19093)

It also fixes a long-standing bug in expiring transitioned objects.
The expiration action was deleting the current version in the case'
of tiered objects instead of adding a delete marker.
---
 cmd/bucket-lifecycle.go | 28 +++++++++++++++++++---------
 cmd/data-scanner.go     | 40 +++++++++++++++++++++++++++++-----------
 2 files changed, 48 insertions(+), 20 deletions(-)

diff --git a/cmd/bucket-lifecycle.go b/cmd/bucket-lifecycle.go
index 22450bb7f..f5270d380 100644
--- a/cmd/bucket-lifecycle.go
+++ b/cmd/bucket-lifecycle.go
@@ -395,12 +395,15 @@ func enqueueTransitionImmediate(obj ObjectInfo, src lcEventSrc) {
 //
 // 1. when a restored (via PostRestoreObject API) object expires.
 // 2. when a transitioned object expires (based on an ILM rule).
-func expireTransitionedObject(ctx context.Context, objectAPI ObjectLayer, oi *ObjectInfo, lcOpts lifecycle.ObjectOpts, lcEvent lifecycle.Event, src lcEventSrc) error {
+func expireTransitionedObject(ctx context.Context, objectAPI ObjectLayer, oi *ObjectInfo, lcEvent lifecycle.Event, src lcEventSrc) error {
 	traceFn := globalLifecycleSys.trace(*oi)
-	var opts ObjectOptions
-	opts.Versioned = globalBucketVersioningSys.PrefixEnabled(oi.Bucket, oi.Name)
-	opts.VersionID = lcOpts.VersionID
-	opts.Expiration = ExpirationOptions{Expire: true}
+	opts := ObjectOptions{
+		Versioned:  globalBucketVersioningSys.PrefixEnabled(oi.Bucket, oi.Name),
+		Expiration: ExpirationOptions{Expire: true},
+	}
+	if lcEvent.Action.DeleteVersioned() {
+		opts.VersionID = oi.VersionID
+	}
 	tags := newLifecycleAuditEvent(src, lcEvent).Tags()
 	if lcEvent.Action.DeleteRestored() {
 		// delete locally restored copy of object or object version
@@ -435,13 +438,13 @@ func expireTransitionedObject(ctx context.Context, objectAPI ObjectLayer, oi *Ob
 	defer auditLogLifecycle(ctx, *oi, ILMExpiry, tags, traceFn)
 
 	eventName := event.ObjectRemovedDelete
-	if lcOpts.DeleteMarker {
+	if oi.DeleteMarker {
 		eventName = event.ObjectRemovedDeleteMarkerCreated
 	}
 	objInfo := ObjectInfo{
 		Name:         oi.Name,
-		VersionID:    lcOpts.VersionID,
-		DeleteMarker: lcOpts.DeleteMarker,
+		VersionID:    oi.VersionID,
+		DeleteMarker: oi.DeleteMarker,
 	}
 	// Notify object deleted event.
 	sendEvent(eventArgs{
@@ -471,7 +474,14 @@ func genTransitionObjName(bucket string) (string, error) {
 // storage specified by the transition ARN, the metadata is left behind on source cluster and original content
 // is moved to the transition tier. Note that in the case of encrypted objects, entire encrypted stream is moved
 // to the transition tier without decrypting or re-encrypting.
-func transitionObject(ctx context.Context, objectAPI ObjectLayer, oi ObjectInfo, lae lcAuditEvent) error {
+func transitionObject(ctx context.Context, objectAPI ObjectLayer, oi ObjectInfo, lae lcAuditEvent) (err error) {
+	defer func() {
+		if err != nil {
+			return
+		}
+		globalScannerMetrics.timeILM(lae.Action)(1)
+	}()
+
 	opts := ObjectOptions{
 		Transition: TransitionOptions{
 			Status: lifecycle.TransitionPending,
diff --git a/cmd/data-scanner.go b/cmd/data-scanner.go
index 8ccd3854b..86ed7e883 100644
--- a/cmd/data-scanner.go
+++ b/cmd/data-scanner.go
@@ -963,15 +963,6 @@ func (i *scannerItem) applyLifecycle(ctx context.Context, o ObjectLayer, oi Obje
 			console.Debugf(applyActionsLogPrefix+" lifecycle: %q Initial scan: %v\n", i.objectPath(), lcEvt.Action)
 		}
 	}
-	defer func() {
-		if lcEvt.Action != lifecycle.NoneAction {
-			numVersions := uint64(1)
-			if lcEvt.Action == lifecycle.DeleteAllVersionsAction {
-				numVersions = uint64(oi.NumVersions)
-			}
-			globalScannerMetrics.timeILM(lcEvt.Action)(numVersions)
-		}
-	}()
 
 	switch lcEvt.Action {
 	// This version doesn't contribute towards sizeS only when it is permanently deleted.
@@ -1228,7 +1219,17 @@ func applyTransitionRule(event lifecycle.Event, src lcEventSrc, obj ObjectInfo)
 }
 
 func applyExpiryOnTransitionedObject(ctx context.Context, objLayer ObjectLayer, obj ObjectInfo, lcEvent lifecycle.Event, src lcEventSrc) bool {
-	if err := expireTransitionedObject(ctx, objLayer, &obj, obj.ToLifecycleOpts(), lcEvent, src); err != nil {
+	var err error
+	defer func() {
+		if err != nil {
+			return
+		}
+		// Note: DeleteAllVersions action is not supported for
+		// transitioned objects
+		globalScannerMetrics.timeILM(lcEvent.Action)(1)
+	}()
+
+	if err = expireTransitionedObject(ctx, objLayer, &obj, lcEvent, src); err != nil {
 		if isErrObjectNotFound(err) || isErrVersionNotFound(err) {
 			return false
 		}
@@ -1255,8 +1256,25 @@ func applyExpiryOnNonTransitionedObjects(ctx context.Context, objLayer ObjectLay
 	if lcEvent.Action.DeleteAll() {
 		opts.DeletePrefix = true
 	}
+	var (
+		dobj ObjectInfo
+		err  error
+	)
+	defer func() {
+		if err != nil {
+			return
+		}
 
-	dobj, err := objLayer.DeleteObject(ctx, obj.Bucket, obj.Name, opts)
+		if lcEvent.Action != lifecycle.NoneAction {
+			numVersions := uint64(1)
+			if lcEvent.Action == lifecycle.DeleteAllVersionsAction {
+				numVersions = uint64(obj.NumVersions)
+			}
+			globalScannerMetrics.timeILM(lcEvent.Action)(numVersions)
+		}
+	}()
+
+	dobj, err = objLayer.DeleteObject(ctx, obj.Bucket, obj.Name, opts)
 	if err != nil {
 		if isErrObjectNotFound(err) || isErrVersionNotFound(err) {
 			return false