Merge branch 'feature/per-repo-endpoint' of https://github.com/brutasse/elasticsearch-cloud-aws into brutasse-feature/per-repo-endpoint

Author: tlrx

README.md

@@ -163,6 +163,8 @@ The following settings are supported:
 
 * `bucket`: The name of the bucket to be used for snapshots. (Mandatory)
 * `region`: The region where bucket is located. Defaults to US Standard
+* `endpoint`: The endpoint to the S3 API. Defaults to AWS's default S3 endpoint. Note that setting a region overrides the endpoint setting.
+* `protocol`: The protocol to use (`http` or `https`). Defaults to `https`.
 * `base_path`: Specifies the path within bucket to repository data. Defaults to root directory.
 * `access_key`: The access key to use for authentication. Defaults to value of `cloud.aws.access_key`.
 * `secret_key`: The secret key to use for authentication. Defaults to value of `cloud.aws.secret_key`.
@@ -289,10 +291,16 @@ repositories:
  remote-bucket:
  bucket: <bucket in other region>
  region: <region>
+external-bucket:
+ bucket: <bucket>
+ access_key: <access key>
+ secret_key: <secret key>
+ endpoint: <endpoint>
+ protocol: <protocol>
 
 ```
 
-Replace all occurrences of `access_key`, `secret_key`, `bucket` and `region` with your settings. Please, note that the test will delete all snapshot/restore related files in the specified buckets.
+Replace all occurrences of `access_key`, `secret_key`, `endpoint`, `protocol`, `bucket` and `region` with your settings. Please, note that the test will delete all snapshot/restore related files in the specified buckets.
 
 To run test:
 

src/main/java/org/elasticsearch/cloud/aws/AwsS3Service.java

@@ -28,7 +28,7 @@
 public interface AwsS3Service extends LifecycleComponent<AwsS3Service> {
  AmazonS3 client();
 
- AmazonS3 client(String region, String account, String key);
+ AmazonS3 client(String endpoint, String protocol, String region, String account, String key);
 
- AmazonS3 client(String region, String account, String key, Integer maxRetries);
+ AmazonS3 client(String endpoint, String protocol, String region, String account, String key, Integer maxRetries);
 }

src/main/java/org/elasticsearch/cloud/aws/InternalAwsS3Service.java

@@ -60,42 +60,43 @@ public synchronized AmazonS3 client() {
  String account = componentSettings.get("access_key", settings.get("cloud.account"));
  String key = componentSettings.get("secret_key", settings.get("cloud.key"));
 
- return getClient(endpoint, account, key, null);
+ return getClient(endpoint, "https", account, key, null);
  }
 
  @Override
- public AmazonS3 client(String region, String account, String key) {
- return client(region, account, key, null);
+ public AmazonS3 client(String endpoint, String protocol, String region, String account, String key) {
+ return client(endpoint, protocol, region, account, key, null);
  }
 
  @Override
- public synchronized AmazonS3 client(String region, String account, String key, Integer maxRetries) {
- String endpoint;
- if (region == null) {
- endpoint = getDefaultEndpoint();
- } else {
+ public synchronized AmazonS3 client(String endpoint, String protocol, String region, String account, String key, Integer maxRetries) {
+ if (region != null && endpoint == null) {
  endpoint = getEndpoint(region);
  logger.debug("using s3 region [{}], with endpoint [{}]", region, endpoint);
+ } else if (endpoint == null) {
+ endpoint = getDefaultEndpoint();
  }
  if (account == null || key == null) {
  account = componentSettings.get("access_key", settings.get("cloud.account"));
  key = componentSettings.get("secret_key", settings.get("cloud.key"));
  }
 
- return getClient(endpoint, account, key, maxRetries);
+ return getClient(endpoint, protocol, account, key, maxRetries);
  }
 
 
- private synchronized AmazonS3 getClient(String endpoint, String account, String key, Integer maxRetries) {
+ private synchronized AmazonS3 getClient(String endpoint, String protocol, String account, String key, Integer maxRetries) {
  Tuple<String, String> clientDescriptor = new Tuple<String, String>(endpoint, account);
  AmazonS3Client client = clients.get(clientDescriptor);
  if (client != null) {
  return client;
  }
 
  ClientConfiguration clientConfiguration = new ClientConfiguration();
- String protocol = componentSettings.get("protocol", "https").toLowerCase();
- protocol = componentSettings.get("s3.protocol", protocol).toLowerCase();
+ if (protocol == null) {
+ protocol = "https";
+ }
+
  if ("http".equals(protocol)) {
  clientConfiguration.setProtocol(Protocol.HTTP);
  } else if ("https".equals(protocol)) {

src/main/java/org/elasticsearch/repositories/s3/S3Repository.java

@@ -79,6 +79,11 @@ public S3Repository(RepositoryName name, RepositorySettings repositorySettings,
  throw new RepositoryException(name.name(), "No bucket defined for s3 gateway");
  }
 
+ String endpoint = repositorySettings.settings().get("endpoint", componentSettings.get("endpoint"));
+ String protocol = componentSettings.get("protocol", "https").toLowerCase();
+ protocol = componentSettings.get("s3.protocol", protocol).toLowerCase();
+ protocol = repositorySettings.settings().get("protocol", protocol);
+
  String region = repositorySettings.settings().get("region", componentSettings.get("region"));
  if (region == null) {
  // Bucket setting is not set - use global region setting
@@ -124,10 +129,10 @@ public S3Repository(RepositoryName name, RepositorySettings repositorySettings,
  this.chunkSize = repositorySettings.settings().getAsBytesSize("chunk_size", componentSettings.getAsBytesSize("chunk_size", new ByteSizeValue(100, ByteSizeUnit.MB)));
  this.compress = repositorySettings.settings().getAsBoolean("compress", componentSettings.getAsBoolean("compress", false));
 
- logger.debug("using bucket [{}], region [{}], chunk_size [{}], server_side_encryption [{}], buffer_size [{}], max_retries [{}]",
- bucket, region, chunkSize, serverSideEncryption, bufferSize, maxRetries);
+ logger.debug("using bucket [{}], region [{}], endpoint [{}], protocol [{}], chunk_size [{}], server_side_encryption [{}], buffer_size [{}], max_retries [{}]",
+ bucket, region, endpoint, protocol, chunkSize, serverSideEncryption, bufferSize, maxRetries);
 
- blobStore = new S3BlobStore(settings, s3Service.client(region, repositorySettings.settings().get("access_key"), repositorySettings.settings().get("secret_key"), maxRetries), bucket, region, serverSideEncryption, bufferSize, maxRetries);
+ blobStore = new S3BlobStore(settings, s3Service.client(endpoint, protocol, region, repositorySettings.settings().get("access_key"), repositorySettings.settings().get("secret_key"), maxRetries), bucket, region, serverSideEncryption, bufferSize, maxRetries);
  String basePath = repositorySettings.settings().get("base_path", null);
  if (Strings.hasLength(basePath)) {
  BlobPath path = new BlobPath();

src/test/java/org/elasticsearch/cloud/aws/TestAwsS3Service.java

@@ -45,13 +45,13 @@ public synchronized AmazonS3 client() {
  }
 
  @Override
- public synchronized AmazonS3 client(String region, String account, String key) {
- return cachedWrapper(super.client(region, account, key));
+ public synchronized AmazonS3 client(String endpoint, String protocol, String region, String account, String key) {
+ return cachedWrapper(super.client(endpoint, protocol, region, account, key));
  }
 
  @Override
- public synchronized AmazonS3 client(String region, String account, String key, Integer maxRetries) {
- return cachedWrapper(super.client(region, account, key, maxRetries));
+ public synchronized AmazonS3 client(String endpoint, String protocol, String region, String account, String key, Integer maxRetries) {
+ return cachedWrapper(super.client(endpoint, protocol, region, account, key, maxRetries));
  }
 
  private AmazonS3 cachedWrapper(AmazonS3 client) {

src/test/java/org/elasticsearch/repositories/s3/AbstractS3SnapshotRestoreTest.java

@@ -192,6 +192,8 @@ public void testEncryption() {
 Settings settings = internalCluster().getInstance(Settings.class);
 Settings bucket = settings.getByPrefix("repositories.s3.");
 AmazonS3 s3Client = internalCluster().getInstance(AwsS3Service.class).client(
+bucket.get("endpoint", settings.get("repositories.s3.endpoint")),
+bucket.get("protocol", settings.get("repositories.s3.protocol")),
 bucket.get("region", settings.get("repositories.s3.region")),
 bucket.get("access_key", settings.get("cloud.aws.access_key")),
 bucket.get("secret_key", settings.get("cloud.aws.secret_key")));
@@ -278,6 +280,22 @@ public void testRepositoryWithCustomCredentials() {
  assertRepositoryIsOperational(client, "test-repo");
  }
 
+ @Test
+ public void testRepositoryWithCustomEndpointProtocol() {
+ Client client = client();
+ Settings bucketSettings = internalCluster().getInstance(Settings.class).getByPrefix("repositories.s3.external-bucket.");
+ logger.info("--> creating s3 repostoriy with endpoint [{}], bucket[{}] and path [{}]", bucketSettings.get("endpoint"), bucketSettings.get("bucket"), basePath);
+ PutRepositoryResponse putRepositoryResponse = client.admin().cluster().preparePutRepository("test-repo")
+ .setType("s3").setSettings(ImmutableSettings.settingsBuilder()
+ .put("protocol", bucketSettings.get("protocol"))
+ .put("endpoint", bucketSettings.get("endpoint"))
+ .put("access_key", bucketSettings.get("access_key"))
+ .put("secret_key", bucketSettings.get("secret_key"))
+ ).get();
+ assertThat(putRepositoryResponse.isAcknowledged(), equalTo(true));
+ assertRepositoryIsOperational(client, "test-repo");
+ }
+
  /**
  * This test verifies that the test configuration is set up in a manner that
  * does not make the test {@link #testRepositoryInRemoteRegion()} pointless.
@@ -430,6 +448,8 @@ public void cleanRepositoryFiles(String basePath) {
  settings.getByPrefix("repositories.s3.remote-bucket.")
  };
  for (Settings bucket : buckets) {
+ String endpoint = bucket.get("endpoint", settings.get("repositories.s3.endpoint"));
+ String protocol = bucket.get("protocol", settings.get("repositories.s3.protocol"));
  String region = bucket.get("region", settings.get("repositories.s3.region"));
  String accessKey = bucket.get("access_key", settings.get("cloud.aws.access_key"));
  String secretKey = bucket.get("secret_key", settings.get("cloud.aws.secret_key"));
@@ -438,7 +458,7 @@ public void cleanRepositoryFiles(String basePath) {
  // We check that settings has been set in elasticsearch.yml integration test file
  // as described in README
  assertThat("Your settings in elasticsearch.yml are incorrects. Check README file.", bucketName, notNullValue());
- AmazonS3 client = internalCluster().getInstance(AwsS3Service.class).client(region, accessKey, secretKey);
+ AmazonS3 client = internalCluster().getInstance(AwsS3Service.class).client(endpoint, protocol, region, accessKey, secretKey);
  try {
  ObjectListing prevListing = null;
  //From http://docs.amazonwebservices.com/AmazonS3/latest/dev/DeletingMultipleObjectsUsingJava.html