Added retry mechanism for S3 connection errors

Closes #95

Author: imotov

README.md

@@ -115,6 +115,7 @@ The following settings are supported:
 * `chunk_size`: Big files can be broken down into chunks during snapshotting if needed. The chunk size can be specified in bytes or by using size value notation, i.e. `1g`, `10m`, `5k`. Defaults to `100m`.
 * `compress`: When set to `true` metadata files are stored in compressed format. This setting doesn't affect index files that are already compressed by default. Defaults to `false`.
 * `server_side_encryption`: When set to `true` files are encrypted on server side using AES256 algorithm. Defaults to `false`.
+* `max_retries`: Number of retries in case of S3 errors. Defaults to `3`.
 
 The S3 repositories are using the same credentials as the rest of the AWS services provided by this plugin (`discovery`).
 See [Generic Configuration](#generic-configuration) for details.

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

@@ -20,15 +20,29 @@
 package org.elasticsearch.cloud.aws;
 
 import org.elasticsearch.common.inject.AbstractModule;
+import org.elasticsearch.common.settings.Settings;
 
 /**
  *
  */
 public class AwsModule extends AbstractModule {
 
+    private final Settings settings;
+
+    public static final String S3_SERVICE_TYPE_KEY = "cloud.aws.s3service.type";
+
+    public AwsModule(Settings settings) {
+        this.settings = settings;
+    }
+
     @Override
     protected void configure() {
-        bind(AwsS3Service.class).asEagerSingleton();
+        bind(AwsS3Service.class).to(getS3ServiceClass(settings)).asEagerSingleton();
         bind(AwsEc2Service.class).asEagerSingleton();
     }
-}
+
+    public static Class<? extends AwsS3Service> getS3ServiceClass(Settings settings) {
+        return settings.getAsClass(S3_SERVICE_TYPE_KEY, InternalAwsS3Service.class);
+    }
+
+}

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

@@ -19,176 +19,14 @@
 
 package org.elasticsearch.cloud.aws;
 
-import java.util.HashMap;
-import java.util.Map;
-
-import com.amazonaws.ClientConfiguration;
-import com.amazonaws.Protocol;
-import com.amazonaws.auth.*;
-import com.amazonaws.internal.StaticCredentialsProvider;
 import com.amazonaws.services.s3.AmazonS3;
-import com.amazonaws.services.s3.AmazonS3Client;
-import org.elasticsearch.ElasticsearchException;
-import org.elasticsearch.ElasticsearchIllegalArgumentException;
-import org.elasticsearch.common.collect.Tuple;
-import org.elasticsearch.common.component.AbstractLifecycleComponent;
-import org.elasticsearch.common.inject.Inject;
-import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.settings.SettingsFilter;
+import org.elasticsearch.common.component.LifecycleComponent;
 
 /**
  *
  */
-public class AwsS3Service extends AbstractLifecycleComponent<AwsS3Service> {
-
-    /**
-     * (acceskey, endpoint) -> client
-     */
-    private Map<Tuple<String, String>, AmazonS3Client> clients = new HashMap<Tuple<String,String>, AmazonS3Client>();
-
-    @Inject
-    public AwsS3Service(Settings settings, SettingsFilter settingsFilter) {
-        super(settings);
-
-        settingsFilter.addFilter(new AwsSettingsFilter());
-    }
-
-    public synchronized AmazonS3 client() {
-        String endpoint = getDefaultEndpoint();
-        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);
-    }
-
-    public synchronized AmazonS3 client(String region, String account, String key) {
-        String endpoint;
-        if (region == null) {
-            endpoint = getDefaultEndpoint();
-        } else {
-            endpoint = getEndpoint(region);
-            logger.debug("using s3 region [{}], with endpoint [{}]", region, endpoint);
-        }
-        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);
-    }
-
-
-    private synchronized AmazonS3 getClient(String endpoint, String account, String key) {
-        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", "http").toLowerCase();
-        if ("http".equals(protocol)) {
-            clientConfiguration.setProtocol(Protocol.HTTP);
-        } else if ("https".equals(protocol)) {
-            clientConfiguration.setProtocol(Protocol.HTTPS);
-        } else {
-            throw new ElasticsearchIllegalArgumentException("No protocol supported [" + protocol + "], can either be [http] or [https]");
-        }
-
-        String proxyHost = componentSettings.get("proxy_host");
-        if (proxyHost != null) {
-            String portString = componentSettings.get("proxy_port", "80");
-            Integer proxyPort;
-            try {
-                proxyPort = Integer.parseInt(portString, 10);
-            } catch (NumberFormatException ex) {
-                throw new ElasticsearchIllegalArgumentException("The configured proxy port value [" + portString + "] is invalid", ex);
-            }
-            clientConfiguration.withProxyHost(proxyHost).setProxyPort(proxyPort);
-        }
-
-        AWSCredentialsProvider credentials;
-
-        if (account == null && key == null) {
-            credentials = new AWSCredentialsProviderChain(
-                    new EnvironmentVariableCredentialsProvider(),
-                    new SystemPropertiesCredentialsProvider(),
-                    new InstanceProfileCredentialsProvider()
-            );
-        } else {
-            credentials = new AWSCredentialsProviderChain(
-                    new StaticCredentialsProvider(new BasicAWSCredentials(account, key))
-            );
-        }
-        client = new AmazonS3Client(credentials, clientConfiguration);
-
-        if (endpoint != null) {
-            client.setEndpoint(endpoint);
-        }
-        clients.put(clientDescriptor, client);
-        return client;
-    }
-
-    private String getDefaultEndpoint() {
-        String endpoint = null;
-        if (componentSettings.get("s3.endpoint") != null) {
-            endpoint = componentSettings.get("s3.endpoint");
-            logger.debug("using explicit s3 endpoint [{}]", endpoint);
-        } else if (componentSettings.get("region") != null) {
-            String region = componentSettings.get("region").toLowerCase();
-            endpoint = getEndpoint(region);
-            logger.debug("using s3 region [{}], with endpoint [{}]", region, endpoint);
-        }
-        return endpoint;
-    }
-
-    private static String getEndpoint(String region) {
-        if ("us-east".equals(region)) {
-            return "s3.amazonaws.com";
-        } else if ("us-east-1".equals(region)) {
-            return "s3.amazonaws.com";
-        } else if ("us-west".equals(region)) {
-            return "s3-us-west-1.amazonaws.com";
-        } else if ("us-west-1".equals(region)) {
-            return "s3-us-west-1.amazonaws.com";
-        } else if ("us-west-2".equals(region)) {
-            return "s3-us-west-2.amazonaws.com";
-        } else if ("ap-southeast".equals(region)) {
-            return "s3-ap-southeast-1.amazonaws.com";
-        } else if ("ap-southeast-1".equals(region)) {
-            return "s3-ap-southeast-1.amazonaws.com";
-        } else if ("ap-southeast-2".equals(region)) {
-            return "s3-ap-southeast-2.amazonaws.com";
-        } else if ("ap-northeast".equals(region)) {
-            return "s3-ap-northeast-1.amazonaws.com";
-        } else if ("ap-northeast-1".equals(region)) {
-            return "s3-ap-northeast-1.amazonaws.com";
-        } else if ("eu-west".equals(region)) {
-            return "s3-eu-west-1.amazonaws.com";
-        } else if ("eu-west-1".equals(region)) {
-            return "s3-eu-west-1.amazonaws.com";
-        } else if ("sa-east".equals(region)) {
-            return "s3-sa-east-1.amazonaws.com";
-        } else if ("sa-east-1".equals(region)) {
-            return "s3-sa-east-1.amazonaws.com";
-        } else {
-            throw new ElasticsearchIllegalArgumentException("No automatic endpoint could be derived from region [" + region + "]");
-        }
-    }
-
-    @Override
-    protected void doStart() throws ElasticsearchException {
-    }
-
-    @Override
-    protected void doStop() throws ElasticsearchException {
-    }
-
-    @Override
-    protected void doClose() throws ElasticsearchException {
-        for (AmazonS3Client client : clients.values()) {
-            client.shutdown();
-        }
-    }
+public interface AwsS3Service extends LifecycleComponent<AwsS3Service> {
+    AmazonS3 client();
 
+    AmazonS3 client(String region, String account, String key);
 }