Add IsRemote method on FileInfo, ObjectInfo (#12209)

Provides a convenient method to know if an object's contents are in its remote
tier.

cmd/api-headers.go

@@ -201,13 +201,11 @@ func setObjectHeaders(w http.ResponseWriter, objInfo ObjectInfo, rs *HTTPRangeSp
 				}
 			}
 		}
-		if objInfo.TransitionStatus == lifecycle.TransitionComplete {
+		if objInfo.IsRemote() {
 			// Check if object is being restored. For more information on x-amz-restore header see
 			// https://docs.aws.amazon.com/AmazonS3/latest/API/API_HeadObject.html#API_HeadObject_ResponseSyntax
-			if onDisk := isRestoredObjectOnDisk(objInfo.UserDefined); !onDisk {
 			w.Header()[xhttp.AmzStorageClass] = []string{objInfo.TransitionTier}
 		}
-		}
 		ruleID, transitionTime := lc.PredictTransitionTime(lifecycle.ObjectOpts{
 			Name:             objInfo.Name,
 			UserTags:         objInfo.UserTags,

cmd/bucket-lifecycle.go

@@ -550,6 +550,24 @@ func putRestoreOpts(bucket, object string, rreq *RestoreObjectRequest, objInfo O
 
 var errRestoreHDRMalformed = fmt.Errorf("x-amz-restore header malformed")
 
+// IsRemote returns true if this object version's contents are in its remote
+// tier.
+func (fi FileInfo) IsRemote() bool {
+	if fi.TransitionStatus != lifecycle.TransitionComplete {
+		return false
+	}
+	return !isRestoredObjectOnDisk(fi.Metadata)
+}
+
+// IsRemote returns true if this object version's contents are in its remote
+// tier.
+func (oi ObjectInfo) IsRemote() bool {
+	if oi.TransitionStatus != lifecycle.TransitionComplete {
+		return false
+	}
+	return !isRestoredObjectOnDisk(oi.UserDefined)
+}
+
 // restoreObjStatus represents a restore-object's status. It can be either
 // ongoing or completed.
 type restoreObjStatus struct {

cmd/bucket-lifecycle_test.go

@@ -23,6 +23,7 @@ import (
 	"time"
 
 	xhttp "github.com/minio/minio/cmd/http"
+	"github.com/minio/minio/pkg/bucket/lifecycle"
 )
 
 // TestParseRestoreObjStatus tests parseRestoreObjStatus
@@ -155,3 +156,61 @@ func TestIsRestoredObjectOnDisk(t *testing.T) {
 		}
 	}
 }
+
+func TestObjectIsRemote(t *testing.T) {
+	fi := newFileInfo("object", 8, 8)
+	fi.Erasure.Index = 1
+	if !fi.IsValid() {
+		t.Fatalf("unable to get xl meta")
+	}
+
+	testCases := []struct {
+		meta   map[string]string
+		remote bool
+	}{
+		{
+			// restore in progress
+			meta: map[string]string{
+				xhttp.AmzRestore: ongoingRestoreObj().String(),
+			},
+			remote: true,
+		},
+		{
+			// restore completed
+			meta: map[string]string{
+				xhttp.AmzRestore: completedRestoreObj(time.Now().Add(time.Hour)).String(),
+			},
+			remote: false,
+		},
+		{
+			// restore completed but expired
+			meta: map[string]string{
+				xhttp.AmzRestore: completedRestoreObj(time.Now().Add(-time.Hour)).String(),
+			},
+			remote: true,
+		},
+		{
+			// restore never initiated
+			meta:   map[string]string{},
+			remote: true,
+		},
+	}
+	for i, tc := range testCases {
+		// Set transition status to complete
+		fi.TransitionStatus = lifecycle.TransitionComplete
+		fi.Metadata = tc.meta
+		if got := fi.IsRemote(); got != tc.remote {
+			t.Fatalf("Test %d.a: expected %v got %v", i+1, tc.remote, got)
+		}
+		oi := fi.ToObjectInfo("bucket", "object")
+		if got := oi.IsRemote(); got != tc.remote {
+			t.Fatalf("Test %d.b: expected %v got %v", i+1, tc.remote, got)
+		}
+	}
+	// Reset transition status; An object that's not transitioned is not remote.
+	fi.TransitionStatus = ""
+	fi.Metadata = nil
+	if got := fi.IsRemote(); got != false {
+		t.Fatalf("Expected object not to be remote but got %v", got)
+	}
+}

cmd/erasure-healing.go

@@ -509,7 +509,7 @@ func (er erasureObjects) healObject(ctx context.Context, bucket string, object s
 
 		// dataDir should be empty when
 		// - transitionStatus is complete and not in restored state
-		if partsMetadata[i].TransitionStatus == lifecycle.TransitionComplete && !isRestoredObjectOnDisk(partsMetadata[i].Metadata) {
+		if partsMetadata[i].IsRemote() {
 			partsMetadata[i].DataDir = ""
 		}
 		// Attempt a rename now from healed data to final location.

cmd/erasure-object.go

@@ -183,9 +183,7 @@ func (er erasureObjects) GetObjectNInfo(ctx context.Context, bucket, object stri
 			ObjInfo: objInfo,
 		}, toObjectErr(errMethodNotAllowed, bucket, object)
 	}
-	if objInfo.TransitionStatus == lifecycle.TransitionComplete {
+	if objInfo.IsRemote() {
-		// If transitioned, stream from transition tier unless object is restored locally or restore date is past.
-		if onDisk := isRestoredObjectOnDisk(objInfo.UserDefined); !onDisk {
 		gr, err := getTransitionedObjectReader(ctx, bucket, object, rs, h, objInfo, opts)
 		if err != nil {
 			return nil, err
@@ -193,7 +191,6 @@ func (er erasureObjects) GetObjectNInfo(ctx context.Context, bucket, object stri
 		unlockOnDefer = false
 		return gr.WithCleanupFuncs(nsUnlocker), nil
 	}
-	}
 
 	fn, off, length, err := NewGetObjectReader(rs, objInfo, opts)
 	if err != nil {