mirror of
https://github.com/honeymoose/OpenSearch.git
synced 2025-02-20 11:54:52 +00:00
Removes the retry mechanism from the S3 blob store (#23952)
Currently, both the Amazon S3 client provides a retry mechanism, and the S3 blob store also attempts retries for failed read/write requests. Both retry mechanisms are controlled by the `repositories.s3.max_retries` setting. However, the S3 blob store retry mechanism is unnecessary because the Amazon S3 client provided by the Amazon SDK already handles retries (with exponential backoff) based on the provided max retry configuration setting (defaults to 3) as long as the request is retryable. Hence, this commit removes the unneeded retry logic in the S3 blob store and the S3OutputStream. Closes #22845
This commit is contained in:
parent
53e3ddf2f0
commit
4f121744bd
@ -25,6 +25,7 @@ import java.util.function.Function;
|
|||||||
import com.amazonaws.ClientConfiguration;
|
import com.amazonaws.ClientConfiguration;
|
||||||
import com.amazonaws.Protocol;
|
import com.amazonaws.Protocol;
|
||||||
import com.amazonaws.services.s3.AmazonS3;
|
import com.amazonaws.services.s3.AmazonS3;
|
||||||
|
import org.elasticsearch.cluster.metadata.RepositoryMetaData;
|
||||||
import org.elasticsearch.common.component.LifecycleComponent;
|
import org.elasticsearch.common.component.LifecycleComponent;
|
||||||
import org.elasticsearch.common.settings.SecureString;
|
import org.elasticsearch.common.settings.SecureString;
|
||||||
import org.elasticsearch.common.settings.Setting;
|
import org.elasticsearch.common.settings.Setting;
|
||||||
@ -159,5 +160,8 @@ interface AwsS3Service extends LifecycleComponent {
|
|||||||
Setting.timeSetting("cloud.aws.s3.read_timeout", AwsS3Service.READ_TIMEOUT, Property.NodeScope, Property.Deprecated);
|
Setting.timeSetting("cloud.aws.s3.read_timeout", AwsS3Service.READ_TIMEOUT, Property.NodeScope, Property.Deprecated);
|
||||||
}
|
}
|
||||||
|
|
||||||
AmazonS3 client(Settings repositorySettings, Integer maxRetries, boolean useThrottleRetries, Boolean pathStyleAccess);
|
/**
|
||||||
|
* Creates an {@code AmazonS3} client from the given repository metadata and node settings.
|
||||||
|
*/
|
||||||
|
AmazonS3 client(RepositoryMetaData metadata, Settings repositorySettings);
|
||||||
}
|
}
|
||||||
|
@ -72,8 +72,8 @@ class DefaultS3OutputStream extends S3OutputStream {
|
|||||||
private int multipartChunks;
|
private int multipartChunks;
|
||||||
private List<PartETag> multiparts;
|
private List<PartETag> multiparts;
|
||||||
|
|
||||||
DefaultS3OutputStream(S3BlobStore blobStore, String bucketName, String blobName, int bufferSizeInBytes, int numberOfRetries, boolean serverSideEncryption) {
|
DefaultS3OutputStream(S3BlobStore blobStore, String bucketName, String blobName, int bufferSizeInBytes, boolean serverSideEncryption) {
|
||||||
super(blobStore, bucketName, blobName, bufferSizeInBytes, numberOfRetries, serverSideEncryption);
|
super(blobStore, bucketName, blobName, bufferSizeInBytes, serverSideEncryption);
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
@ -106,19 +106,10 @@ class DefaultS3OutputStream extends S3OutputStream {
|
|||||||
*/
|
*/
|
||||||
private void upload(byte[] bytes, int off, int len) throws IOException {
|
private void upload(byte[] bytes, int off, int len) throws IOException {
|
||||||
try (ByteArrayInputStream is = new ByteArrayInputStream(bytes, off, len)) {
|
try (ByteArrayInputStream is = new ByteArrayInputStream(bytes, off, len)) {
|
||||||
int retry = 0;
|
try {
|
||||||
while (retry <= getNumberOfRetries()) {
|
doUpload(getBlobStore(), getBucketName(), getBlobName(), is, len, isServerSideEncryption());
|
||||||
try {
|
} catch (AmazonClientException e) {
|
||||||
doUpload(getBlobStore(), getBucketName(), getBlobName(), is, len, isServerSideEncryption());
|
throw new IOException("Unable to upload object " + getBlobName(), e);
|
||||||
break;
|
|
||||||
} catch (AmazonClientException e) {
|
|
||||||
if (getBlobStore().shouldRetry(e) && retry < getNumberOfRetries()) {
|
|
||||||
is.reset();
|
|
||||||
retry++;
|
|
||||||
} else {
|
|
||||||
throw new IOException("Unable to upload object " + getBlobName(), e);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -131,10 +122,9 @@ class DefaultS3OutputStream extends S3OutputStream {
|
|||||||
}
|
}
|
||||||
md.setContentLength(length);
|
md.setContentLength(length);
|
||||||
|
|
||||||
InputStream inputStream = is;
|
|
||||||
|
|
||||||
// We try to compute a MD5 while reading it
|
// We try to compute a MD5 while reading it
|
||||||
MessageDigest messageDigest;
|
MessageDigest messageDigest;
|
||||||
|
InputStream inputStream;
|
||||||
try {
|
try {
|
||||||
messageDigest = MessageDigest.getInstance("MD5");
|
messageDigest = MessageDigest.getInstance("MD5");
|
||||||
inputStream = new DigestInputStream(is, messageDigest);
|
inputStream = new DigestInputStream(is, messageDigest);
|
||||||
@ -159,20 +149,11 @@ class DefaultS3OutputStream extends S3OutputStream {
|
|||||||
}
|
}
|
||||||
|
|
||||||
private void initializeMultipart() {
|
private void initializeMultipart() {
|
||||||
int retry = 0;
|
while (multipartId == null) {
|
||||||
while ((retry <= getNumberOfRetries()) && (multipartId == null)) {
|
multipartId = doInitialize(getBlobStore(), getBucketName(), getBlobName(), isServerSideEncryption());
|
||||||
try {
|
if (multipartId != null) {
|
||||||
multipartId = doInitialize(getBlobStore(), getBucketName(), getBlobName(), isServerSideEncryption());
|
multipartChunks = 1;
|
||||||
if (multipartId != null) {
|
multiparts = new ArrayList<>();
|
||||||
multipartChunks = 1;
|
|
||||||
multiparts = new ArrayList<>();
|
|
||||||
}
|
|
||||||
} catch (AmazonClientException e) {
|
|
||||||
if (getBlobStore().shouldRetry(e) && retry < getNumberOfRetries()) {
|
|
||||||
retry++;
|
|
||||||
} else {
|
|
||||||
throw e;
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -193,22 +174,13 @@ class DefaultS3OutputStream extends S3OutputStream {
|
|||||||
|
|
||||||
private void uploadMultipart(byte[] bytes, int off, int len, boolean lastPart) throws IOException {
|
private void uploadMultipart(byte[] bytes, int off, int len, boolean lastPart) throws IOException {
|
||||||
try (ByteArrayInputStream is = new ByteArrayInputStream(bytes, off, len)) {
|
try (ByteArrayInputStream is = new ByteArrayInputStream(bytes, off, len)) {
|
||||||
int retry = 0;
|
try {
|
||||||
while (retry <= getNumberOfRetries()) {
|
PartETag partETag = doUploadMultipart(getBlobStore(), getBucketName(), getBlobName(), multipartId, is, len, lastPart);
|
||||||
try {
|
multiparts.add(partETag);
|
||||||
PartETag partETag = doUploadMultipart(getBlobStore(), getBucketName(), getBlobName(), multipartId, is, len, lastPart);
|
multipartChunks++;
|
||||||
multiparts.add(partETag);
|
} catch (AmazonClientException e) {
|
||||||
multipartChunks++;
|
abortMultipart();
|
||||||
return;
|
throw e;
|
||||||
} catch (AmazonClientException e) {
|
|
||||||
if (getBlobStore().shouldRetry(e) && retry < getNumberOfRetries()) {
|
|
||||||
is.reset();
|
|
||||||
retry++;
|
|
||||||
} else {
|
|
||||||
abortMultipart();
|
|
||||||
throw e;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -230,20 +202,13 @@ class DefaultS3OutputStream extends S3OutputStream {
|
|||||||
}
|
}
|
||||||
|
|
||||||
private void completeMultipart() {
|
private void completeMultipart() {
|
||||||
int retry = 0;
|
try {
|
||||||
while (retry <= getNumberOfRetries()) {
|
doCompleteMultipart(getBlobStore(), getBucketName(), getBlobName(), multipartId, multiparts);
|
||||||
try {
|
multipartId = null;
|
||||||
doCompleteMultipart(getBlobStore(), getBucketName(), getBlobName(), multipartId, multiparts);
|
return;
|
||||||
multipartId = null;
|
} catch (AmazonClientException e) {
|
||||||
return;
|
abortMultipart();
|
||||||
} catch (AmazonClientException e) {
|
throw e;
|
||||||
if (getBlobStore().shouldRetry(e) && retry < getNumberOfRetries()) {
|
|
||||||
retry++;
|
|
||||||
} else {
|
|
||||||
abortMultipart();
|
|
||||||
throw e;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -36,6 +36,7 @@ import com.amazonaws.services.s3.AmazonS3Client;
|
|||||||
import com.amazonaws.services.s3.S3ClientOptions;
|
import com.amazonaws.services.s3.S3ClientOptions;
|
||||||
import org.apache.logging.log4j.Logger;
|
import org.apache.logging.log4j.Logger;
|
||||||
import org.elasticsearch.ElasticsearchException;
|
import org.elasticsearch.ElasticsearchException;
|
||||||
|
import org.elasticsearch.cluster.metadata.RepositoryMetaData;
|
||||||
import org.elasticsearch.common.Strings;
|
import org.elasticsearch.common.Strings;
|
||||||
import org.elasticsearch.common.collect.Tuple;
|
import org.elasticsearch.common.collect.Tuple;
|
||||||
import org.elasticsearch.common.component.AbstractLifecycleComponent;
|
import org.elasticsearch.common.component.AbstractLifecycleComponent;
|
||||||
@ -45,6 +46,8 @@ import org.elasticsearch.common.settings.Setting;
|
|||||||
import org.elasticsearch.common.settings.Settings;
|
import org.elasticsearch.common.settings.Settings;
|
||||||
import org.elasticsearch.common.unit.TimeValue;
|
import org.elasticsearch.common.unit.TimeValue;
|
||||||
|
|
||||||
|
import static org.elasticsearch.repositories.s3.S3Repository.getValue;
|
||||||
|
|
||||||
class InternalAwsS3Service extends AbstractLifecycleComponent implements AwsS3Service {
|
class InternalAwsS3Service extends AbstractLifecycleComponent implements AwsS3Service {
|
||||||
|
|
||||||
// pkg private for tests
|
// pkg private for tests
|
||||||
@ -60,8 +63,7 @@ class InternalAwsS3Service extends AbstractLifecycleComponent implements AwsS3Se
|
|||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public synchronized AmazonS3 client(Settings repositorySettings, Integer maxRetries,
|
public synchronized AmazonS3 client(RepositoryMetaData metadata, Settings repositorySettings) {
|
||||||
boolean useThrottleRetries, Boolean pathStyleAccess) {
|
|
||||||
String clientName = CLIENT_NAME.get(repositorySettings);
|
String clientName = CLIENT_NAME.get(repositorySettings);
|
||||||
String foundEndpoint = findEndpoint(logger, repositorySettings, settings, clientName);
|
String foundEndpoint = findEndpoint(logger, repositorySettings, settings, clientName);
|
||||||
|
|
||||||
@ -73,6 +75,26 @@ class InternalAwsS3Service extends AbstractLifecycleComponent implements AwsS3Se
|
|||||||
return client;
|
return client;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
Integer maxRetries = getValue(metadata.settings(), settings,
|
||||||
|
S3Repository.Repository.MAX_RETRIES_SETTING,
|
||||||
|
S3Repository.Repositories.MAX_RETRIES_SETTING);
|
||||||
|
boolean useThrottleRetries = getValue(metadata.settings(), settings,
|
||||||
|
S3Repository.Repository.USE_THROTTLE_RETRIES_SETTING,
|
||||||
|
S3Repository.Repositories.USE_THROTTLE_RETRIES_SETTING);
|
||||||
|
// If the user defined a path style access setting, we rely on it,
|
||||||
|
// otherwise we use the default value set by the SDK
|
||||||
|
Boolean pathStyleAccess = null;
|
||||||
|
if (S3Repository.Repository.PATH_STYLE_ACCESS_SETTING.exists(metadata.settings()) ||
|
||||||
|
S3Repository.Repositories.PATH_STYLE_ACCESS_SETTING.exists(settings)) {
|
||||||
|
pathStyleAccess = getValue(metadata.settings(), settings,
|
||||||
|
S3Repository.Repository.PATH_STYLE_ACCESS_SETTING,
|
||||||
|
S3Repository.Repositories.PATH_STYLE_ACCESS_SETTING);
|
||||||
|
}
|
||||||
|
|
||||||
|
logger.debug("creating S3 client with client_name [{}], endpoint [{}], max_retries [{}], " +
|
||||||
|
"use_throttle_retries [{}], path_style_access [{}]",
|
||||||
|
clientName, foundEndpoint, maxRetries, useThrottleRetries, pathStyleAccess);
|
||||||
|
|
||||||
client = new AmazonS3Client(
|
client = new AmazonS3Client(
|
||||||
credentials,
|
credentials,
|
||||||
buildConfiguration(logger, repositorySettings, settings, clientName, maxRetries, foundEndpoint, useThrottleRetries));
|
buildConfiguration(logger, repositorySettings, settings, clientName, maxRetries, foundEndpoint, useThrottleRetries));
|
||||||
@ -187,7 +209,7 @@ class InternalAwsS3Service extends AbstractLifecycleComponent implements AwsS3Se
|
|||||||
// no repository setting, just use global setting
|
// no repository setting, just use global setting
|
||||||
return globalSetting.get(globalSettings);
|
return globalSetting.get(globalSettings);
|
||||||
} else {
|
} else {
|
||||||
return S3Repository.getValue(repositorySettings, globalSettings, repositorySetting, globalSetting);
|
return getValue(repositorySettings, globalSettings, repositorySetting, globalSetting);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -73,25 +73,17 @@ class S3BlobContainer extends AbstractBlobContainer {
|
|||||||
|
|
||||||
@Override
|
@Override
|
||||||
public InputStream readBlob(String blobName) throws IOException {
|
public InputStream readBlob(String blobName) throws IOException {
|
||||||
int retry = 0;
|
try {
|
||||||
while (retry <= blobStore.numberOfRetries()) {
|
S3Object s3Object = SocketAccess.doPrivileged(() -> blobStore.client().getObject(blobStore.bucket(), buildKey(blobName)));
|
||||||
try {
|
return s3Object.getObjectContent();
|
||||||
S3Object s3Object = SocketAccess.doPrivileged(() -> blobStore.client().getObject(blobStore.bucket(), buildKey(blobName)));
|
} catch (AmazonClientException e) {
|
||||||
return s3Object.getObjectContent();
|
if (e instanceof AmazonS3Exception) {
|
||||||
} catch (AmazonClientException e) {
|
if (404 == ((AmazonS3Exception) e).getStatusCode()) {
|
||||||
if (blobStore.shouldRetry(e) && (retry < blobStore.numberOfRetries())) {
|
throw new NoSuchFileException("Blob object [" + blobName + "] not found: " + e.getMessage());
|
||||||
retry++;
|
|
||||||
} else {
|
|
||||||
if (e instanceof AmazonS3Exception) {
|
|
||||||
if (404 == ((AmazonS3Exception) e).getStatusCode()) {
|
|
||||||
throw new NoSuchFileException("Blob object [" + blobName + "] not found: " + e.getMessage());
|
|
||||||
}
|
|
||||||
}
|
|
||||||
throw e;
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
throw e;
|
||||||
}
|
}
|
||||||
throw new BlobStoreException("retries exhausted while attempting to access blob object [name:" + blobName + ", bucket:" + blobStore.bucket() + "]");
|
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
@ -120,7 +112,7 @@ class S3BlobContainer extends AbstractBlobContainer {
|
|||||||
private OutputStream createOutput(final String blobName) throws IOException {
|
private OutputStream createOutput(final String blobName) throws IOException {
|
||||||
// UploadS3OutputStream does buffering & retry logic internally
|
// UploadS3OutputStream does buffering & retry logic internally
|
||||||
return new DefaultS3OutputStream(blobStore, blobStore.bucket(), buildKey(blobName),
|
return new DefaultS3OutputStream(blobStore, blobStore.bucket(), buildKey(blobName),
|
||||||
blobStore.bufferSizeInBytes(), blobStore.numberOfRetries(), blobStore.serverSideEncryption());
|
blobStore.bufferSizeInBytes(), blobStore.serverSideEncryption());
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
@ -51,21 +51,18 @@ class S3BlobStore extends AbstractComponent implements BlobStore {
|
|||||||
|
|
||||||
private final boolean serverSideEncryption;
|
private final boolean serverSideEncryption;
|
||||||
|
|
||||||
private final int numberOfRetries;
|
|
||||||
|
|
||||||
private final CannedAccessControlList cannedACL;
|
private final CannedAccessControlList cannedACL;
|
||||||
|
|
||||||
private final StorageClass storageClass;
|
private final StorageClass storageClass;
|
||||||
|
|
||||||
S3BlobStore(Settings settings, AmazonS3 client, String bucket, boolean serverSideEncryption,
|
S3BlobStore(Settings settings, AmazonS3 client, String bucket, boolean serverSideEncryption,
|
||||||
ByteSizeValue bufferSize, int maxRetries, String cannedACL, String storageClass) {
|
ByteSizeValue bufferSize, String cannedACL, String storageClass) {
|
||||||
super(settings);
|
super(settings);
|
||||||
this.client = client;
|
this.client = client;
|
||||||
this.bucket = bucket;
|
this.bucket = bucket;
|
||||||
this.serverSideEncryption = serverSideEncryption;
|
this.serverSideEncryption = serverSideEncryption;
|
||||||
this.bufferSize = bufferSize;
|
this.bufferSize = bufferSize;
|
||||||
this.cannedACL = initCannedACL(cannedACL);
|
this.cannedACL = initCannedACL(cannedACL);
|
||||||
this.numberOfRetries = maxRetries;
|
|
||||||
this.storageClass = initStorageClass(storageClass);
|
this.storageClass = initStorageClass(storageClass);
|
||||||
|
|
||||||
// Note: the method client.doesBucketExist() may return 'true' is the bucket exists
|
// Note: the method client.doesBucketExist() may return 'true' is the bucket exists
|
||||||
@ -102,10 +99,6 @@ class S3BlobStore extends AbstractComponent implements BlobStore {
|
|||||||
return bufferSize.bytesAsInt();
|
return bufferSize.bytesAsInt();
|
||||||
}
|
}
|
||||||
|
|
||||||
public int numberOfRetries() {
|
|
||||||
return numberOfRetries;
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public BlobContainer blobContainer(BlobPath path) {
|
public BlobContainer blobContainer(BlobPath path) {
|
||||||
return new S3BlobContainer(path, this);
|
return new S3BlobContainer(path, this);
|
||||||
|
@ -39,7 +39,6 @@ abstract class S3OutputStream extends OutputStream {
|
|||||||
private S3BlobStore blobStore;
|
private S3BlobStore blobStore;
|
||||||
private String bucketName;
|
private String bucketName;
|
||||||
private String blobName;
|
private String blobName;
|
||||||
private int numberOfRetries;
|
|
||||||
private boolean serverSideEncryption;
|
private boolean serverSideEncryption;
|
||||||
|
|
||||||
private byte[] buffer;
|
private byte[] buffer;
|
||||||
@ -48,11 +47,10 @@ abstract class S3OutputStream extends OutputStream {
|
|||||||
|
|
||||||
private int flushCount = 0;
|
private int flushCount = 0;
|
||||||
|
|
||||||
S3OutputStream(S3BlobStore blobStore, String bucketName, String blobName, int bufferSizeInBytes, int numberOfRetries, boolean serverSideEncryption) {
|
S3OutputStream(S3BlobStore blobStore, String bucketName, String blobName, int bufferSizeInBytes, boolean serverSideEncryption) {
|
||||||
this.blobStore = blobStore;
|
this.blobStore = blobStore;
|
||||||
this.bucketName = bucketName;
|
this.bucketName = bucketName;
|
||||||
this.blobName = blobName;
|
this.blobName = blobName;
|
||||||
this.numberOfRetries = numberOfRetries;
|
|
||||||
this.serverSideEncryption = serverSideEncryption;
|
this.serverSideEncryption = serverSideEncryption;
|
||||||
|
|
||||||
if (bufferSizeInBytes < MULTIPART_MIN_SIZE.getBytes()) {
|
if (bufferSizeInBytes < MULTIPART_MIN_SIZE.getBytes()) {
|
||||||
@ -107,10 +105,6 @@ abstract class S3OutputStream extends OutputStream {
|
|||||||
return buffer.length;
|
return buffer.length;
|
||||||
}
|
}
|
||||||
|
|
||||||
public int getNumberOfRetries() {
|
|
||||||
return numberOfRetries;
|
|
||||||
}
|
|
||||||
|
|
||||||
public boolean isServerSideEncryption() {
|
public boolean isServerSideEncryption() {
|
||||||
return serverSideEncryption;
|
return serverSideEncryption;
|
||||||
}
|
}
|
||||||
|
@ -311,8 +311,6 @@ class S3Repository extends BlobStoreRepository {
|
|||||||
|
|
||||||
boolean serverSideEncryption = getValue(metadata.settings(), settings, Repository.SERVER_SIDE_ENCRYPTION_SETTING, Repositories.SERVER_SIDE_ENCRYPTION_SETTING);
|
boolean serverSideEncryption = getValue(metadata.settings(), settings, Repository.SERVER_SIDE_ENCRYPTION_SETTING, Repositories.SERVER_SIDE_ENCRYPTION_SETTING);
|
||||||
ByteSizeValue bufferSize = getValue(metadata.settings(), settings, Repository.BUFFER_SIZE_SETTING, Repositories.BUFFER_SIZE_SETTING);
|
ByteSizeValue bufferSize = getValue(metadata.settings(), settings, Repository.BUFFER_SIZE_SETTING, Repositories.BUFFER_SIZE_SETTING);
|
||||||
Integer maxRetries = getValue(metadata.settings(), settings, Repository.MAX_RETRIES_SETTING, Repositories.MAX_RETRIES_SETTING);
|
|
||||||
boolean useThrottleRetries = getValue(metadata.settings(), settings, Repository.USE_THROTTLE_RETRIES_SETTING, Repositories.USE_THROTTLE_RETRIES_SETTING);
|
|
||||||
this.chunkSize = getValue(metadata.settings(), settings, Repository.CHUNK_SIZE_SETTING, Repositories.CHUNK_SIZE_SETTING);
|
this.chunkSize = getValue(metadata.settings(), settings, Repository.CHUNK_SIZE_SETTING, Repositories.CHUNK_SIZE_SETTING);
|
||||||
this.compress = getValue(metadata.settings(), settings, Repository.COMPRESS_SETTING, Repositories.COMPRESS_SETTING);
|
this.compress = getValue(metadata.settings(), settings, Repository.COMPRESS_SETTING, Repositories.COMPRESS_SETTING);
|
||||||
|
|
||||||
@ -326,21 +324,12 @@ class S3Repository extends BlobStoreRepository {
|
|||||||
String storageClass = getValue(metadata.settings(), settings, Repository.STORAGE_CLASS_SETTING, Repositories.STORAGE_CLASS_SETTING);
|
String storageClass = getValue(metadata.settings(), settings, Repository.STORAGE_CLASS_SETTING, Repositories.STORAGE_CLASS_SETTING);
|
||||||
String cannedACL = getValue(metadata.settings(), settings, Repository.CANNED_ACL_SETTING, Repositories.CANNED_ACL_SETTING);
|
String cannedACL = getValue(metadata.settings(), settings, Repository.CANNED_ACL_SETTING, Repositories.CANNED_ACL_SETTING);
|
||||||
|
|
||||||
// If the user defined a path style access setting, we rely on it otherwise we use the default
|
|
||||||
// value set by the SDK
|
|
||||||
Boolean pathStyleAccess = null;
|
|
||||||
if (Repository.PATH_STYLE_ACCESS_SETTING.exists(metadata.settings()) ||
|
|
||||||
Repositories.PATH_STYLE_ACCESS_SETTING.exists(settings)) {
|
|
||||||
pathStyleAccess = getValue(metadata.settings(), settings, Repository.PATH_STYLE_ACCESS_SETTING, Repositories.PATH_STYLE_ACCESS_SETTING);
|
|
||||||
}
|
|
||||||
|
|
||||||
logger.debug("using bucket [{}], chunk_size [{}], server_side_encryption [{}], " +
|
logger.debug("using bucket [{}], chunk_size [{}], server_side_encryption [{}], " +
|
||||||
"buffer_size [{}], max_retries [{}], use_throttle_retries [{}], cannedACL [{}], storageClass [{}], path_style_access [{}]",
|
"buffer_size [{}], cannedACL [{}], storageClass [{}]",
|
||||||
bucket, chunkSize, serverSideEncryption, bufferSize, maxRetries, useThrottleRetries, cannedACL,
|
bucket, chunkSize, serverSideEncryption, bufferSize, cannedACL, storageClass);
|
||||||
storageClass, pathStyleAccess);
|
|
||||||
|
|
||||||
AmazonS3 client = s3Service.client(metadata.settings(), maxRetries, useThrottleRetries, pathStyleAccess);
|
AmazonS3 client = s3Service.client(metadata, metadata.settings());
|
||||||
blobStore = new S3BlobStore(settings, client, bucket, serverSideEncryption, bufferSize, maxRetries, cannedACL, storageClass);
|
blobStore = new S3BlobStore(settings, client, bucket, serverSideEncryption, bufferSize, cannedACL, storageClass);
|
||||||
|
|
||||||
String basePath = getValue(metadata.settings(), settings, Repository.BASE_PATH_SETTING, Repositories.BASE_PATH_SETTING);
|
String basePath = getValue(metadata.settings(), settings, Repository.BASE_PATH_SETTING, Repositories.BASE_PATH_SETTING);
|
||||||
if (Strings.hasLength(basePath)) {
|
if (Strings.hasLength(basePath)) {
|
||||||
|
@ -31,6 +31,7 @@ import org.elasticsearch.action.admin.cluster.snapshots.restore.RestoreSnapshotR
|
|||||||
import org.elasticsearch.client.Client;
|
import org.elasticsearch.client.Client;
|
||||||
import org.elasticsearch.client.ClusterAdminClient;
|
import org.elasticsearch.client.ClusterAdminClient;
|
||||||
import org.elasticsearch.cluster.ClusterState;
|
import org.elasticsearch.cluster.ClusterState;
|
||||||
|
import org.elasticsearch.cluster.metadata.RepositoryMetaData;
|
||||||
import org.elasticsearch.common.settings.Settings;
|
import org.elasticsearch.common.settings.Settings;
|
||||||
import org.elasticsearch.repositories.RepositoryMissingException;
|
import org.elasticsearch.repositories.RepositoryMissingException;
|
||||||
import org.elasticsearch.repositories.RepositoryVerificationException;
|
import org.elasticsearch.repositories.RepositoryVerificationException;
|
||||||
@ -165,6 +166,7 @@ public abstract class AbstractS3SnapshotRestoreTest extends AbstractAwsTestCase
|
|||||||
.put(S3Repository.Repository.BASE_PATH_SETTING.getKey(), basePath)
|
.put(S3Repository.Repository.BASE_PATH_SETTING.getKey(), basePath)
|
||||||
.put(S3Repository.Repository.CHUNK_SIZE_SETTING.getKey(), randomIntBetween(1000, 10000))
|
.put(S3Repository.Repository.CHUNK_SIZE_SETTING.getKey(), randomIntBetween(1000, 10000))
|
||||||
.put(S3Repository.Repository.SERVER_SIDE_ENCRYPTION_SETTING.getKey(), true)
|
.put(S3Repository.Repository.SERVER_SIDE_ENCRYPTION_SETTING.getKey(), true)
|
||||||
|
.put(S3Repository.Repository.USE_THROTTLE_RETRIES_SETTING.getKey(), randomBoolean())
|
||||||
.build();
|
.build();
|
||||||
|
|
||||||
PutRepositoryResponse putRepositoryResponse = client.admin().cluster().preparePutRepository("test-repo")
|
PutRepositoryResponse putRepositoryResponse = client.admin().cluster().preparePutRepository("test-repo")
|
||||||
@ -194,7 +196,8 @@ public abstract class AbstractS3SnapshotRestoreTest extends AbstractAwsTestCase
|
|||||||
|
|
||||||
Settings settings = internalCluster().getInstance(Settings.class);
|
Settings settings = internalCluster().getInstance(Settings.class);
|
||||||
Settings bucket = settings.getByPrefix("repositories.s3.");
|
Settings bucket = settings.getByPrefix("repositories.s3.");
|
||||||
AmazonS3 s3Client = internalCluster().getInstance(AwsS3Service.class).client(repositorySettings, null, randomBoolean(), null);
|
RepositoryMetaData metadata = new RepositoryMetaData("test-repo", "fs", Settings.EMPTY);
|
||||||
|
AmazonS3 s3Client = internalCluster().getInstance(AwsS3Service.class).client(metadata, repositorySettings);
|
||||||
|
|
||||||
String bucketName = bucket.get("bucket");
|
String bucketName = bucket.get("bucket");
|
||||||
logger.info("--> verify encryption for bucket [{}], prefix [{}]", bucketName, basePath);
|
logger.info("--> verify encryption for bucket [{}], prefix [{}]", bucketName, basePath);
|
||||||
@ -462,7 +465,9 @@ public abstract class AbstractS3SnapshotRestoreTest extends AbstractAwsTestCase
|
|||||||
// We check that settings has been set in elasticsearch.yml integration test file
|
// We check that settings has been set in elasticsearch.yml integration test file
|
||||||
// as described in README
|
// as described in README
|
||||||
assertThat("Your settings in elasticsearch.yml are incorrects. Check README file.", bucketName, notNullValue());
|
assertThat("Your settings in elasticsearch.yml are incorrects. Check README file.", bucketName, notNullValue());
|
||||||
AmazonS3 client = internalCluster().getInstance(AwsS3Service.class).client(Settings.EMPTY, null, randomBoolean(), null);
|
RepositoryMetaData metadata = new RepositoryMetaData("test-repo", "fs", Settings.EMPTY);
|
||||||
|
AmazonS3 client = internalCluster().getInstance(AwsS3Service.class).client(metadata,
|
||||||
|
Settings.builder().put(S3Repository.Repository.USE_THROTTLE_RETRIES_SETTING.getKey(), randomBoolean()).build());
|
||||||
try {
|
try {
|
||||||
ObjectListing prevListing = null;
|
ObjectListing prevListing = null;
|
||||||
//From http://docs.amazonwebservices.com/AmazonS3/latest/dev/DeletingMultipleObjectsUsingJava.html
|
//From http://docs.amazonwebservices.com/AmazonS3/latest/dev/DeletingMultipleObjectsUsingJava.html
|
||||||
|
@ -26,9 +26,6 @@ import com.amazonaws.auth.AWSCredentialsProvider;
|
|||||||
import org.elasticsearch.common.settings.MockSecureSettings;
|
import org.elasticsearch.common.settings.MockSecureSettings;
|
||||||
import org.elasticsearch.common.settings.Setting;
|
import org.elasticsearch.common.settings.Setting;
|
||||||
import org.elasticsearch.common.settings.Settings;
|
import org.elasticsearch.common.settings.Settings;
|
||||||
import org.elasticsearch.repositories.s3.AwsS3Service;
|
|
||||||
import org.elasticsearch.repositories.s3.InternalAwsS3Service;
|
|
||||||
import org.elasticsearch.repositories.s3.S3Repository;
|
|
||||||
import org.elasticsearch.test.ESTestCase;
|
import org.elasticsearch.test.ESTestCase;
|
||||||
|
|
||||||
import static org.hamcrest.Matchers.instanceOf;
|
import static org.hamcrest.Matchers.instanceOf;
|
||||||
|
@ -42,7 +42,7 @@ public class MockDefaultS3OutputStream extends DefaultS3OutputStream {
|
|||||||
private int numberOfUploadRequests = 0;
|
private int numberOfUploadRequests = 0;
|
||||||
|
|
||||||
public MockDefaultS3OutputStream(int bufferSizeInBytes) {
|
public MockDefaultS3OutputStream(int bufferSizeInBytes) {
|
||||||
super(null, "test-bucket", "test-blobname", bufferSizeInBytes, 3, false);
|
super(null, "test-bucket", "test-blobname", bufferSizeInBytes, false);
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
@ -34,6 +34,6 @@ public class S3BlobStoreContainerTests extends ESBlobStoreContainerTestCase {
|
|||||||
String bucket = randomAlphaOfLength(randomIntBetween(1, 10)).toLowerCase(Locale.ROOT);
|
String bucket = randomAlphaOfLength(randomIntBetween(1, 10)).toLowerCase(Locale.ROOT);
|
||||||
|
|
||||||
return new S3BlobStore(Settings.EMPTY, client, bucket, false,
|
return new S3BlobStore(Settings.EMPTY, client, bucket, false,
|
||||||
new ByteSizeValue(10, ByteSizeUnit.MB), 5, "public-read-write", "standard");
|
new ByteSizeValue(10, ByteSizeUnit.MB), "public-read-write", "standard");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -60,8 +60,7 @@ public class S3RepositoryTests extends ESTestCase {
|
|||||||
@Override
|
@Override
|
||||||
protected void doClose() {}
|
protected void doClose() {}
|
||||||
@Override
|
@Override
|
||||||
public AmazonS3 client(Settings repositorySettings, Integer maxRetries,
|
public AmazonS3 client(RepositoryMetaData metadata, Settings settings) {
|
||||||
boolean useThrottleRetries, Boolean pathStyleAccess) {
|
|
||||||
return new DummyS3Client();
|
return new DummyS3Client();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -23,6 +23,7 @@ import java.util.IdentityHashMap;
|
|||||||
|
|
||||||
import com.amazonaws.services.s3.AmazonS3;
|
import com.amazonaws.services.s3.AmazonS3;
|
||||||
import org.elasticsearch.ElasticsearchException;
|
import org.elasticsearch.ElasticsearchException;
|
||||||
|
import org.elasticsearch.cluster.metadata.RepositoryMetaData;
|
||||||
import org.elasticsearch.common.settings.Settings;
|
import org.elasticsearch.common.settings.Settings;
|
||||||
|
|
||||||
public class TestAwsS3Service extends InternalAwsS3Service {
|
public class TestAwsS3Service extends InternalAwsS3Service {
|
||||||
@ -33,7 +34,7 @@ public class TestAwsS3Service extends InternalAwsS3Service {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
IdentityHashMap<AmazonS3, TestAmazonS3> clients = new IdentityHashMap<AmazonS3, TestAmazonS3>();
|
IdentityHashMap<AmazonS3, TestAmazonS3> clients = new IdentityHashMap<>();
|
||||||
|
|
||||||
public TestAwsS3Service(Settings settings) {
|
public TestAwsS3Service(Settings settings) {
|
||||||
super(settings);
|
super(settings);
|
||||||
@ -41,9 +42,8 @@ public class TestAwsS3Service extends InternalAwsS3Service {
|
|||||||
|
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public synchronized AmazonS3 client(Settings repositorySettings, Integer maxRetries,
|
public synchronized AmazonS3 client(RepositoryMetaData metadata, Settings repositorySettings) {
|
||||||
boolean useThrottleRetries, Boolean pathStyleAccess) {
|
return cachedWrapper(super.client(metadata, repositorySettings));
|
||||||
return cachedWrapper(super.client(repositorySettings, maxRetries, useThrottleRetries, pathStyleAccess));
|
|
||||||
}
|
}
|
||||||
|
|
||||||
private AmazonS3 cachedWrapper(AmazonS3 client) {
|
private AmazonS3 cachedWrapper(AmazonS3 client) {
|
||||||
|
Loading…
x
Reference in New Issue
Block a user