Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.fs.Options;
import org.apache.hadoop.fs.s3a.api.S3ObjectStorageClassFilter;
import org.apache.hadoop.security.ssl.DelegatingSSLSocketFactory;

import java.time.Duration;
Expand Down Expand Up @@ -1519,6 +1520,18 @@ 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";

/**
* Default value of Read Restored Glacier objects config.
*/
public static final String DEFAULT_READ_RESTORED_GLACIER_OBJECTS =
S3ObjectStorageClassFilter.READ_ALL.toString();

/**
* The bucket region header.
*/
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathFilter;
import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.fs.s3a.api.S3ObjectStorageClassFilter;
import org.apache.hadoop.fs.s3a.impl.AbstractStoreOperation;
import org.apache.hadoop.fs.s3a.impl.ListingOperationCallbacks;
import org.apache.hadoop.fs.s3a.impl.StoreContext;
Expand Down Expand Up @@ -76,6 +77,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();
Expand All @@ -87,6 +89,7 @@ public Listing(ListingOperationCallbacks listingOperationCallbacks,
super(storeContext);
this.listingOperationCallbacks = listingOperationCallbacks;
this.isCSEEnabled = storeContext.isCSEEnabled();
this.s3ObjectStorageClassFilter = storeContext.getS3ObjectsStorageClassFilter();
}

/**
Expand Down Expand Up @@ -462,7 +465,10 @@ 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 based on the config fs.s3a.glacier.read.restored.objects
if (s3ObjectStorageClassFilter.getFilter().apply(s3Object) &&
acceptor.accept(keyPath, s3Object) &&
filter.accept(keyPath)) {
S3AFileStatus status = createFileStatus(keyPath, s3Object,
listingOperationCallbacks.getDefaultBlockSize(keyPath),
getStoreContext().getUsername(),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -113,6 +113,7 @@
import org.apache.hadoop.fs.Options;
import org.apache.hadoop.fs.impl.OpenFileParameters;
import org.apache.hadoop.fs.permission.FsAction;
import org.apache.hadoop.fs.s3a.api.S3ObjectStorageClassFilter;
import org.apache.hadoop.fs.s3a.audit.AuditSpanS3A;
import org.apache.hadoop.fs.s3a.auth.SignerManager;
import org.apache.hadoop.fs.s3a.auth.delegation.DelegationOperations;
Expand Down Expand Up @@ -444,6 +445,12 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
*/
private boolean isCSEEnabled;

/**
* {@link S3ObjectStorageClassFilter} will filter the S3 files based on the
* {@code fs.s3a.glacier.read.restored.objects} configuration.
*/
private S3ObjectStorageClassFilter s3ObjectStorageClassFilter;

/**
* Bucket AccessPoint.
*/
Expand Down Expand Up @@ -585,6 +592,18 @@ public void initialize(URI name, Configuration originalConf)

s3aInternals = createS3AInternals();

try {
s3ObjectStorageClassFilter = Optional.of(conf.getTrimmed(READ_RESTORED_GLACIER_OBJECTS,
DEFAULT_READ_RESTORED_GLACIER_OBJECTS))
.map(String::toUpperCase)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(Locale.ROOT)

.map(S3ObjectStorageClassFilter::valueOf).get();
} catch (IllegalArgumentException e) {
LOG.warn("Invalid value for the config {} is set. Valid values are:" +
"READ_ALL, SKIP_ALL_GLACIER, READ_RESTORED_GLACIER_OBJECTS. Defaulting to READ_ALL",
READ_RESTORED_GLACIER_OBJECTS);
s3ObjectStorageClassFilter = S3ObjectStorageClassFilter.READ_ALL;
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

lets fall to the default. maybe pull the conf.getTrimmed() out of the try {} so it's value can be printed too.

FWIW in #6789 I'm doing a getEnumSet() which is case independent too.

}

// look for encryption data
// DT Bindings may override this
setEncryptionSecrets(
Expand Down Expand Up @@ -5686,6 +5705,7 @@ public StoreContext createStoreContext() {
.setContextAccessors(new ContextAccessorsImpl())
.setAuditor(getAuditor())
.setEnableCSE(isCSEEnabled)
.setS3ObjectStorageClassFilter(s3ObjectStorageClassFilter)
.build();
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,9 @@
import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
import software.amazon.awssdk.services.s3.model.ObjectIdentifier;
import software.amazon.awssdk.services.s3.model.PutObjectRequest;
import software.amazon.awssdk.services.s3.model.RestoreObjectRequest;
import software.amazon.awssdk.services.s3.model.StorageClass;
import software.amazon.awssdk.services.s3.model.Tier;
import software.amazon.awssdk.services.s3.model.UploadPartRequest;

import org.apache.hadoop.classification.InterfaceAudience;
Expand Down Expand Up @@ -251,4 +253,15 @@ ListObjectsV2Request.Builder newListObjectsV2RequestBuilder(String key,
DeleteObjectsRequest.Builder newBulkDeleteRequestBuilder(
List<ObjectIdentifier> keysToDelete);

/**
* Create a request builder to initiate a restore of Glacier object.
* @param key object to restore
* @param tier glacier retrieval tier at which the restore will be processed.
* @param expirationDays lifetime of the active restored copy in days.
* @return the request builder
*/
RestoreObjectRequest.Builder newRestoreObjectRequestBuilder(String key,
Tier tier,
int expirationDays);

}
Original file line number Diff line number Diff line change
@@ -0,0 +1,97 @@
/*
* 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.api;

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;

import org.apache.hadoop.fs.s3a.S3AFileSystem;
import org.apache.hadoop.util.Sets;


/**
* <pre>
* {@link S3ObjectStorageClassFilter} will filter the S3 files based on the
* {@code fs.s3a.glacier.read.restored.objects} configuration set in {@link S3AFileSystem}
* The config can have 3 values:
* {@code READ_ALL}: Retrieval of Glacier files will fail with InvalidObjectStateException:
* The operation is not valid for the object's storage class.
* {@code 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.
* {@code 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.
* </pre>
*/
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;
}

/**
* Checks if the s3 object is not an object with a storage class of glacier/deep_archive.
* @param object s3 object
* @return if the s3 object is not an object with a storage class of glacier/deep_archive
*/
private static boolean isNotGlacierObject(S3Object object) {
return !GLACIER_STORAGE_CLASSES.contains(object.storageClass());
}

/**
* Checks if the s3 object is an object with a storage class of glacier/deep_archive.
* @param object s3 object
* @return if the s3 object is an object with a storage class of glacier/deep_archive
*/
private static boolean isGlacierObject(S3Object object) {
return GLACIER_STORAGE_CLASSES.contains(object.storageClass());
}

/**
* Checks if the s3 object is completely restored.
* @param object s3 object
* @return if the s3 object is completely restored
*/
private static boolean isCompletedRestoredObject(S3Object object) {
if(isGlacierObject(object)) {
return object.restoreStatus() != null && !object.restoreStatus().isRestoreInProgress();
}
return true;
}

/**
* Returns the filter function set as part of the enum definition
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

needs a trailing .

* @return the filter function set as part of the enum definition
*/
public Function<S3Object, Boolean> getFilter() {
return filter;
}

}
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
import software.amazon.awssdk.services.s3.model.DeleteObjectsRequest;
import software.amazon.awssdk.services.s3.model.GetObjectRequest;
import software.amazon.awssdk.services.s3.model.CreateMultipartUploadRequest;
import software.amazon.awssdk.services.s3.model.GlacierJobParameters;
import software.amazon.awssdk.services.s3.model.HeadBucketRequest;
import software.amazon.awssdk.services.s3.model.HeadObjectRequest;
import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
Expand All @@ -42,9 +43,13 @@
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.RestoreObjectRequest;
import software.amazon.awssdk.services.s3.model.RestoreRequest;
import software.amazon.awssdk.services.s3.model.ServerSideEncryption;
import software.amazon.awssdk.services.s3.model.StorageClass;
import software.amazon.awssdk.services.s3.model.Tier;
import software.amazon.awssdk.services.s3.model.UploadPartRequest;
import software.amazon.awssdk.utils.Md5Utils;
import org.apache.hadoop.util.Preconditions;
Expand Down Expand Up @@ -609,6 +614,8 @@ public ListObjectsV2Request.Builder newListObjectsV2RequestBuilder(
final ListObjectsV2Request.Builder requestBuilder = ListObjectsV2Request.builder()
.bucket(bucket)
.maxKeys(maxKeys)
// Optional Attribute to get the Restored Status of the Glacier Objects
.optionalObjectAttributes(OptionalObjectAttributes.RESTORE_STATUS)
.prefix(key);

if (delimiter != null) {
Expand All @@ -632,6 +639,21 @@ public DeleteObjectsRequest.Builder newBulkDeleteRequestBuilder(
.delete(d -> d.objects(keysToDelete).quiet(!LOG.isTraceEnabled())));
}

@Override
public RestoreObjectRequest.Builder newRestoreObjectRequestBuilder(String key,
Tier tier,
int expirationDays) {
return prepareRequest(RestoreObjectRequest
.builder()
.bucket(bucket)
.key(key)
.restoreRequest(RestoreRequest
.builder()
.days(expirationDays)
.glacierJobParameters(GlacierJobParameters.builder().tier(tier).build())
.build()));
}

@Override
public void setEncryptionSecrets(final EncryptionSecrets secrets) {
encryptionSecrets = secrets;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -26,13 +26,13 @@
import java.util.concurrent.ExecutorService;

import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.ListeningExecutorService;

import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.MoreExecutors;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.s3a.api.RequestFactory;
import org.apache.hadoop.fs.s3a.api.S3ObjectStorageClassFilter;
import org.apache.hadoop.fs.s3a.audit.AuditSpanS3A;
import org.apache.hadoop.fs.s3a.Invoker;
import org.apache.hadoop.fs.s3a.S3AFileStatus;
Expand Down Expand Up @@ -117,6 +117,8 @@ public class StoreContext implements ActiveThreadSpanSource<AuditSpan> {
/** Is client side encryption enabled? */
private final boolean isCSEEnabled;

private final S3ObjectStorageClassFilter s3ObjectStorageClassFilter;
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: javadocs


/**
* Instantiate.
*/
Expand All @@ -137,7 +139,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;
Expand All @@ -158,6 +161,7 @@ public class StoreContext implements ActiveThreadSpanSource<AuditSpan> {
this.contextAccessors = contextAccessors;
this.auditor = auditor;
this.isCSEEnabled = isCSEEnabled;
this.s3ObjectStorageClassFilter = s3ObjectStorageClassFilter;
}

public URI getFsURI() {
Expand Down Expand Up @@ -411,4 +415,13 @@ public RequestFactory getRequestFactory() {
public boolean isCSEEnabled() {
return isCSEEnabled;
}

/**
* Return the S3ObjectStorageClassFilter object for S3A,
* whose value is set according to the config {@code fs.s3a.glacier.read.restored.objects}.
* @return {@link S3ObjectStorageClassFilter} object
*/
public S3ObjectStorageClassFilter getS3ObjectsStorageClassFilter() {
return s3ObjectStorageClassFilter;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import org.apache.hadoop.fs.s3a.Invoker;
import org.apache.hadoop.fs.s3a.S3AInputPolicy;
import org.apache.hadoop.fs.s3a.S3AStorageStatistics;
import org.apache.hadoop.fs.s3a.api.S3ObjectStorageClassFilter;
import org.apache.hadoop.fs.s3a.audit.AuditSpanS3A;
import org.apache.hadoop.fs.s3a.statistics.S3AStatisticsContext;
import org.apache.hadoop.fs.store.audit.AuditSpanSource;
Expand Down Expand Up @@ -69,6 +70,8 @@ public class StoreContextBuilder {

private boolean isCSEEnabled;

private S3ObjectStorageClassFilter s3ObjectStorageClassFilter;

public StoreContextBuilder setFsURI(final URI fsURI) {
this.fsURI = fsURI;
return this;
Expand Down Expand Up @@ -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,
Expand All @@ -192,6 +201,7 @@ public StoreContext build() {
useListV1,
contextAccessors,
auditor,
isCSEEnabled);
isCSEEnabled,
s3ObjectStorageClassFilter);
}
}
Loading