HADOOP-14837: Glacier read restored objects support

Author: Bhavay Pahuja

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java

@@ -1495,6 +1495,12 @@ private Constants() {
    */
   public static final int DEFAULT_PREFETCH_MAX_BLOCKS_COUNT = 4;
 
+  /**
+   * Read Restored Glacier objects config.
+   * Value = {@value}
+   */
+  public static final String READ_RESTORED_GLACIER_OBJECTS = "fs.s3a.glacier.read.restored.objects";
+
   /**
    * The bucket region header.
    */

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Listing.java

@@ -76,6 +76,7 @@ public class Listing extends AbstractStoreOperation {
 
   private static final Logger LOG = S3AFileSystem.LOG;
   private final boolean isCSEEnabled;
+  private final S3ObjectStorageClassFilter s3ObjectStorageClassFilter;
 
   static final FileStatusAcceptor ACCEPT_ALL_BUT_S3N =
       new AcceptAllButS3nDirs();
@@ -87,6 +88,7 @@ public Listing(ListingOperationCallbacks listingOperationCallbacks,
     super(storeContext);
     this.listingOperationCallbacks = listingOperationCallbacks;
     this.isCSEEnabled = storeContext.isCSEEnabled();
+    this.s3ObjectStorageClassFilter = storeContext.s3ObjectsStorageClassFilter();
   }
 
   /**
@@ -143,7 +145,8 @@ public FileStatusListingIterator createFileStatusListingIterator(
     return new FileStatusListingIterator(
         createObjectListingIterator(listPath, request, span),
         filter,
-        acceptor);
+        acceptor
+        );
   }
 
   /**
@@ -307,6 +310,30 @@ interface FileStatusAcceptor {
      * @return true if the status is accepted else false
      */
     boolean accept(FileStatus status);
+
+    /**
+     * Resolve a list of file status acceptors into a single FileStatusAcceptor
+     * @param acceptors List of file status acceptors
+     * @return combined FileStatusAcceptor
+     */
+    static FileStatusAcceptor resolveFileStatusAcceptors(List<FileStatusAcceptor> acceptors) {
+      return new FileStatusAcceptor() {
+        @Override
+        public boolean accept(Path keyPath, S3Object s3Object) {
+          return acceptors.stream().map(acceptor -> acceptor.accept(keyPath, s3Object)).reduce(true, Boolean::logicalAnd);
+        }
+
+        @Override
+        public boolean accept(Path keyPath, String commonPrefix) {
+          return acceptors.stream().map(acceptor -> acceptor.accept(keyPath, commonPrefix)).reduce(true, Boolean::logicalAnd);
+        }
+
+        @Override
+        public boolean accept(FileStatus status) {
+          return acceptors.stream().map(acceptor -> acceptor.accept(status)).reduce(true, Boolean::logicalAnd);
+        }
+      };
+    }
   }
 
   /**
@@ -462,7 +489,8 @@ private boolean buildNextStatusBatch(S3ListResult objects) {
           LOG.debug("{}: {}", keyPath, stringify(s3Object));
         }
         // Skip over keys that are ourselves and old S3N _$folder$ files
-        if (acceptor.accept(keyPath, s3Object) && filter.accept(keyPath)) {
+        // Handle Glacier Storage Class objects based on the config fs.s3a.glacier.read.restored.objects value set
+        if ( s3ObjectStorageClassFilter.getFilter().apply(s3Object) &&  acceptor.accept(keyPath, s3Object) && filter.accept(keyPath)) {
           S3AFileStatus status = createFileStatus(keyPath, s3Object,
                   listingOperationCallbacks.getDefaultBlockSize(keyPath),
                   getStoreContext().getUsername(),

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java

@@ -145,6 +145,7 @@
 import org.apache.hadoop.fs.s3a.impl.StatusProbeEnum;
 import org.apache.hadoop.fs.s3a.impl.StoreContext;
 import org.apache.hadoop.fs.s3a.impl.StoreContextBuilder;
+import org.apache.hadoop.fs.s3a.Listing.AcceptAllButSelfAndS3nDirs;
 import org.apache.hadoop.fs.s3a.prefetch.S3APrefetchingInputStream;
 import org.apache.hadoop.fs.s3a.tools.MarkerToolOperations;
 import org.apache.hadoop.fs.s3a.tools.MarkerToolOperationsImpl;
@@ -445,6 +446,8 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
    */
   private boolean isCSEEnabled;
 
+  private S3ObjectStorageClassFilter s3ObjectStorageClassFilter;
+
   /**
    * Bucket AccessPoint.
    */
@@ -581,6 +584,12 @@ public void initialize(URI name, Configuration originalConf)
 
       s3aInternals = createS3AInternals();
 
+      s3ObjectStorageClassFilter = Optional.ofNullable(conf.get(READ_RESTORED_GLACIER_OBJECTS))
+          .map(String::trim)
+          .map(String::toUpperCase)
+          .map(S3ObjectStorageClassFilter::valueOf)
+          .orElse(S3ObjectStorageClassFilter.READ_ALL);
+
       // look for encryption data
       // DT Bindings may override this
       setEncryptionSecrets(
@@ -2477,7 +2486,7 @@ public RemoteIterator<S3ALocatedFileStatus> listFilesAndDirectoryMarkers(
           true,
           includeSelf
               ? Listing.ACCEPT_ALL_BUT_S3N
-              : new Listing.AcceptAllButSelfAndS3nDirs(path),
+              : new AcceptAllButSelfAndS3nDirs(path),
           status
       );
     }
@@ -3861,7 +3870,8 @@ public S3AFileStatus probePathStatus(final Path path,
     @Override
     public RemoteIterator<S3ALocatedFileStatus> listFilesIterator(final Path path,
         final boolean recursive) throws IOException {
-      return S3AFileSystem.this.innerListFiles(path, recursive, Listing.ACCEPT_ALL_BUT_S3N, null);
+      return S3AFileSystem.this.innerListFiles(path, recursive,
+              Listing.ACCEPT_ALL_BUT_S3N, null);
     }
   }
 
@@ -5197,7 +5207,7 @@ public RemoteIterator<S3ALocatedFileStatus> listFilesAndEmptyDirectories(
    *
    * @param f path
    * @param recursive recursive listing?
-   * @param acceptor file status filter
+   * @param acceptor file status filters
    * @param status optional status of path to list.
    * @return an iterator over the listing.
    * @throws IOException failure
@@ -5224,7 +5234,7 @@ private RemoteIterator<S3ALocatedFileStatus> innerListFiles(
               listing.getListFilesAssumingDir(path,
                       recursive,
                       acceptor,
-                  getActiveAuditSpan());
+                      getActiveAuditSpan());
       // If there are no list entries present, we
       // fallback to file existence check as the path
       // can be a file or empty directory.
@@ -5769,6 +5779,7 @@ public StoreContext createStoreContext() {
         .setContextAccessors(new ContextAccessorsImpl())
         .setAuditor(getAuditor())
         .setEnableCSE(isCSEEnabled)
+        .setS3ObjectStorageClassFilter(s3ObjectStorageClassFilter)
         .build();
   }
 

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3ObjectStorageClassFilter.java

@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.s3a;
+
+import org.apache.hadoop.thirdparty.com.google.common.collect.Sets;
+import java.util.Set;
+import java.util.function.Function;
+import software.amazon.awssdk.services.s3.model.ObjectStorageClass;
+import software.amazon.awssdk.services.s3.model.S3Object;
+
+
+/**
+ * S3ObjectStorageClassFilter will filter the S3 files based on the fs.s3a.glacier.read.restored.objects configuration set in S3AFileSystem
+ * The config can have 3 values:
+ * READ_ALL: This would conform to the current default behavior of not taking into account the storage classes retrieved from S3. This will be done to keep the current behavior for the customers and not changing the experience for them.
+ * SKIP_ALL_GLACIER: If this value is set then this will ignore any S3 Objects which are tagged with Glacier storage classes and retrieve the others.
+ * READ_RESTORED_GLACIER_OBJECTS: If this value is set then restored status of the Glacier object will be checked, if restored the objects would be read like normal S3 objects else they will be ignored as the objects would not have been retrieved from the S3 Glacier.
+ */
+public enum S3ObjectStorageClassFilter {
+  READ_ALL(o -> true),
+  SKIP_ALL_GLACIER(S3ObjectStorageClassFilter::isNotGlacierObject),
+  READ_RESTORED_GLACIER_OBJECTS(S3ObjectStorageClassFilter::isCompletedRestoredObject);
+
+  private static final Set<ObjectStorageClass> GLACIER_STORAGE_CLASSES = Sets.newHashSet(ObjectStorageClass.GLACIER, ObjectStorageClass.DEEP_ARCHIVE);
+
+  private final Function<S3Object, Boolean> filter;
+
+  S3ObjectStorageClassFilter(Function<S3Object, Boolean> filter) {
+    this.filter = filter;
+  }
+
+  private static boolean isNotGlacierObject(S3Object object) {
+    return !GLACIER_STORAGE_CLASSES.contains(object.storageClass());
+  }
+
+  private static boolean isGlacierObject(S3Object object) {
+    return GLACIER_STORAGE_CLASSES.contains(object.storageClass());
+  }
+
+  private static boolean isCompletedRestoredObject(S3Object object) {
+    if(isGlacierObject(object) ) {
+      return object.restoreStatus() != null && !object.restoreStatus().isRestoreInProgress();
+    }
+    return true;
+  }
+
+  public Function<S3Object, Boolean> getFilter() {
+    return filter;
+  }
+
+}

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/RequestFactoryImpl.java

@@ -42,6 +42,7 @@
 import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
 import software.amazon.awssdk.services.s3.model.MetadataDirective;
 import software.amazon.awssdk.services.s3.model.ObjectIdentifier;
+import software.amazon.awssdk.services.s3.model.OptionalObjectAttributes;
 import software.amazon.awssdk.services.s3.model.PutObjectRequest;
 import software.amazon.awssdk.services.s3.model.SelectObjectContentRequest;
 import software.amazon.awssdk.services.s3.model.ServerSideEncryption;
@@ -624,6 +625,7 @@ public ListObjectsV2Request.Builder newListObjectsV2RequestBuilder(
     final ListObjectsV2Request.Builder requestBuilder = ListObjectsV2Request.builder()
         .bucket(bucket)
         .maxKeys(maxKeys)
+        .optionalObjectAttributes(OptionalObjectAttributes.RESTORE_STATUS) // Optional Attribute to get the Restored Status of the Glacier Objects
         .prefix(key);
 
     if (delimiter != null) {

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/StoreContext.java

@@ -25,6 +25,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
 
+import org.apache.hadoop.fs.s3a.S3ObjectStorageClassFilter;
 import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.ListeningExecutorService;
 
 import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.MoreExecutors;
@@ -117,6 +118,8 @@ public class StoreContext implements ActiveThreadSpanSource<AuditSpan> {
   /** Is client side encryption enabled? */
   private final boolean isCSEEnabled;
 
+  private final S3ObjectStorageClassFilter s3ObjectStorageClassFilter;
+
   /**
    * Instantiate.
    */
@@ -137,7 +140,8 @@ public class StoreContext implements ActiveThreadSpanSource<AuditSpan> {
       final boolean useListV1,
       final ContextAccessors contextAccessors,
       final AuditSpanSource<AuditSpanS3A> auditor,
-      final boolean isCSEEnabled) {
+      final boolean isCSEEnabled,
+      final S3ObjectStorageClassFilter s3ObjectStorageClassFilter) {
     this.fsURI = fsURI;
     this.bucket = bucket;
     this.configuration = configuration;
@@ -158,6 +162,7 @@ public class StoreContext implements ActiveThreadSpanSource<AuditSpan> {
     this.contextAccessors = contextAccessors;
     this.auditor = auditor;
     this.isCSEEnabled = isCSEEnabled;
+    this.s3ObjectStorageClassFilter = s3ObjectStorageClassFilter;
   }
 
   public URI getFsURI() {
@@ -411,4 +416,8 @@ public RequestFactory getRequestFactory() {
   public boolean isCSEEnabled() {
     return isCSEEnabled;
   }
+
+  public S3ObjectStorageClassFilter s3ObjectsStorageClassFilter() {
+    return s3ObjectStorageClassFilter;
+  }
 }

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/StoreContextBuilder.java

@@ -22,6 +22,7 @@
 import java.util.concurrent.ExecutorService;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.s3a.S3ObjectStorageClassFilter;
 import org.apache.hadoop.fs.s3a.Invoker;
 import org.apache.hadoop.fs.s3a.S3AInputPolicy;
 import org.apache.hadoop.fs.s3a.S3AStorageStatistics;
@@ -69,6 +70,8 @@ public class StoreContextBuilder {
 
   private boolean isCSEEnabled;
 
+  private S3ObjectStorageClassFilter s3ObjectStorageClassFilter;
+
   public StoreContextBuilder setFsURI(final URI fsURI) {
     this.fsURI = fsURI;
     return this;
@@ -175,6 +178,12 @@ public StoreContextBuilder setEnableCSE(
     return this;
   }
 
+  public StoreContextBuilder setS3ObjectStorageClassFilter(
+      S3ObjectStorageClassFilter value) {
+    s3ObjectStorageClassFilter = value;
+    return this;
+  }
+
   public StoreContext build() {
     return new StoreContext(fsURI,
         bucket,
@@ -192,6 +201,7 @@ public StoreContext build() {
         useListV1,
         contextAccessors,
         auditor,
-        isCSEEnabled);
+        isCSEEnabled,
+        s3ObjectStorageClassFilter);
   }
 }

hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/index.md

@@ -1261,6 +1261,20 @@ The switch to turn S3A auditing on or off.
   </description>
 </property>
 
+<!--
+The switch to control how S3A handles glacier storage classes.
+-->
+<property>
+<name>fs.s3a.glacier.read.restored.objects</name>
+<value>READ_ALL</value>
+<description>
+  The config can have 3 values:
+
+  * READ_ALL: This would conform to the current default behavior of not taking into account the storage classes retrieved from S3. This will be done to keep the current behavior (i.e failing for an unrestored glacier class file) for the customers and not changing the experience for them.
+  * SKIP_ALL_GLACIER: If this value is set then this will ignore any S3 Objects which are tagged with Glacier storage classes and retrieve the others.
+  * READ_RESTORED_GLACIER_OBJECTS: If this value is set then restored status of the Glacier object will be checked, if restored the objects would be read like normal S3 objects else they will be ignored as the objects would not have been retrieved from the S3 Glacier.
+</description>
+</property>
 ```
 ## <a name="retry_and_recovery"></a>Retry and Recovery