Merge pull request #21956 from alexshadow007/aws_read_timeout

Add setting to set read timeout for EC2 discovery and S3 repository plugins

docs/plugins/discovery-ec2.asciidoc

@@ -131,6 +131,17 @@ If you are using a compatible EC2 service, they might be using an older API to s
 You can set your compatible signer API using `cloud.aws.signer` (or `cloud.aws.ec2.signer`)
 with the right signer to use.
 
+===== Read timeout
+
+Read timeout determines the amount of time to wait for data to be transferred over an established,
+open connection before the connection is timed out. Defaults to AWS SDK default value (`50 s`).
+It can be configured with `cloud.aws.read_timeout` (or `cloud.aws.ec2.read_timeout`) setting:
+
+[source, yaml]
+----
+cloud.aws.read_timeout: 30s
+----
+
 [[discovery-ec2-discovery]]
 ==== EC2 Discovery
 
@@ -179,7 +190,6 @@ The following are a list of settings (prefixed with `discovery.ec2`) that can fu
     How long the list of hosts is cached to prevent further requests to the AWS API.
     Defaults to `10s`.
 
-
 [IMPORTANT]
 .Binding the network host
 ==============================================

docs/plugins/repository-s3.asciidoc

@@ -139,6 +139,17 @@ signer to use.
 If you are using a compatible S3 service which do not support Version 4 signing process, you may need to
 use `S3SignerType`, which is Signature Version 2.
 
+===== Read timeout
+
+Read timeout determines the amount of time to wait for data to be transferred over an established,
+open connection before the connection is timed out. Defaults to AWS SDK default value (`50s`).
+It can be configured with `cloud.aws.read_timeout` (or `cloud.aws.s3.read_timeout`) setting:
+
+[source, yaml]
+----
+cloud.aws.read_timeout: 30s
+----
+
 [[repository-s3-repository]]
 ==== S3 Repository
 

plugins/discovery-ec2/src/main/java/org/elasticsearch/cloud/aws/AwsEc2Service.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.cloud.aws;
 
+import com.amazonaws.ClientConfiguration;
 import com.amazonaws.Protocol;
 import com.amazonaws.services.ec2.AmazonEC2;
 import org.elasticsearch.common.settings.Setting;
@@ -80,6 +81,11 @@ public interface AwsEc2Service {
      */
     Setting<String> REGION_SETTING =
         new Setting<>("cloud.aws.region", "", s -> s.toLowerCase(Locale.ROOT), Property.NodeScope, Property.Shared);
+    /**
+     * cloud.aws.read_timeout: Socket read timeout. Shared with repository-s3 plugin
+     */
+    Setting<TimeValue> READ_TIMEOUT = Setting.timeSetting("cloud.aws.read_timeout",
+        TimeValue.timeValueMillis(ClientConfiguration.DEFAULT_SOCKET_TIMEOUT), Property.NodeScope, Property.Shared);
 
     /**
      * Defines specific ec2 settings starting with cloud.aws.ec2.
@@ -146,6 +152,12 @@ public interface AwsEc2Service {
          * cloud.aws.ec2.endpoint: Endpoint. If not set, endpoint will be guessed based on region setting.
          */
         Setting<String> ENDPOINT_SETTING = Setting.simpleString("cloud.aws.ec2.endpoint", Property.NodeScope);
+        /**
+         * cloud.aws.ec2.read_timeout: Socket read timeout. Defaults to cloud.aws.read_timeout
+         * @see AwsEc2Service#READ_TIMEOUT
+         */
+        Setting<TimeValue> READ_TIMEOUT =
+            Setting.timeSetting("cloud.aws.ec2.read_timeout", AwsEc2Service.READ_TIMEOUT, Property.NodeScope);
     }
 
     /**

plugins/discovery-ec2/src/main/java/org/elasticsearch/cloud/aws/AwsEc2ServiceImpl.java

@@ -125,6 +125,7 @@ public class AwsEc2ServiceImpl extends AbstractComponent implements AwsEc2Servic
             10,
             false);
         clientConfiguration.setRetryPolicy(retryPolicy);
+        clientConfiguration.setSocketTimeout((int) CLOUD_EC2.READ_TIMEOUT.get(settings).millis());
 
         return clientConfiguration;
     }

plugins/discovery-ec2/src/main/java/org/elasticsearch/plugin/discovery/ec2/Ec2DiscoveryPlugin.java

@@ -135,6 +135,7 @@ public class Ec2DiscoveryPlugin extends Plugin implements DiscoveryPlugin, Close
         AwsEc2Service.PROXY_PASSWORD_SETTING,
         AwsEc2Service.SIGNER_SETTING,
         AwsEc2Service.REGION_SETTING,
+        AwsEc2Service.READ_TIMEOUT,
         // Register EC2 specific settings: cloud.aws.ec2
         AwsEc2Service.CLOUD_EC2.KEY_SETTING,
         AwsEc2Service.CLOUD_EC2.SECRET_SETTING,
@@ -146,6 +147,7 @@ public class Ec2DiscoveryPlugin extends Plugin implements DiscoveryPlugin, Close
         AwsEc2Service.CLOUD_EC2.SIGNER_SETTING,
         AwsEc2Service.CLOUD_EC2.REGION_SETTING,
         AwsEc2Service.CLOUD_EC2.ENDPOINT_SETTING,
+        AwsEc2Service.CLOUD_EC2.READ_TIMEOUT,
         // Register EC2 discovery settings: discovery.ec2
         AwsEc2Service.DISCOVERY_EC2.HOST_TYPE_SETTING,
         AwsEc2Service.DISCOVERY_EC2.ANY_GROUP_SETTING,

plugins/discovery-ec2/src/test/java/org/elasticsearch/cloud/aws/AwsEc2ServiceImplTests.java

@@ -72,7 +72,8 @@ public class AwsEc2ServiceImplTests extends ESTestCase {
     }
 
     public void testAWSDefaultConfiguration() {
-        launchAWSConfigurationTest(Settings.EMPTY, Protocol.HTTPS, null, -1, null, null, null);
+        launchAWSConfigurationTest(Settings.EMPTY, Protocol.HTTPS, null, -1, null, null, null,
+            ClientConfiguration.DEFAULT_SOCKET_TIMEOUT);
     }
 
     public void testAWSConfigurationWithAwsSettings() {
@@ -83,9 +84,10 @@ public class AwsEc2ServiceImplTests extends ESTestCase {
             .put(AwsEc2Service.PROXY_USERNAME_SETTING.getKey(), "aws_proxy_username")
             .put(AwsEc2Service.PROXY_PASSWORD_SETTING.getKey(), "aws_proxy_password")
             .put(AwsEc2Service.SIGNER_SETTING.getKey(), "AWS3SignerType")
+            .put(AwsEc2Service.READ_TIMEOUT.getKey(), "10s")
             .build();
         launchAWSConfigurationTest(settings, Protocol.HTTP, "aws_proxy_host", 8080, "aws_proxy_username", "aws_proxy_password",
-            "AWS3SignerType");
+            "AWS3SignerType", 10000);
     }
 
     public void testAWSConfigurationWithAwsAndEc2Settings() {
@@ -102,9 +104,10 @@ public class AwsEc2ServiceImplTests extends ESTestCase {
             .put(AwsEc2Service.CLOUD_EC2.PROXY_USERNAME_SETTING.getKey(), "ec2_proxy_username")
             .put(AwsEc2Service.CLOUD_EC2.PROXY_PASSWORD_SETTING.getKey(), "ec2_proxy_password")
             .put(AwsEc2Service.CLOUD_EC2.SIGNER_SETTING.getKey(), "NoOpSignerType")
+            .put(AwsEc2Service.CLOUD_EC2.READ_TIMEOUT.getKey(), "10s")
             .build();
         launchAWSConfigurationTest(settings, Protocol.HTTPS, "ec2_proxy_host", 8081, "ec2_proxy_username", "ec2_proxy_password",
-            "NoOpSignerType");
+            "NoOpSignerType", 10000);
     }
 
     protected void launchAWSConfigurationTest(Settings settings,
@@ -113,7 +116,8 @@ public class AwsEc2ServiceImplTests extends ESTestCase {
                                               int expectedProxyPort,
                                               String expectedProxyUsername,
                                               String expectedProxyPassword,
-                                              String expectedSigner) {
+                                              String expectedSigner,
+                                              int expectedReadTimeout) {
         ClientConfiguration configuration = AwsEc2ServiceImpl.buildConfiguration(logger, settings);
 
         assertThat(configuration.getResponseMetadataCacheSize(), is(0));
@@ -123,6 +127,7 @@ public class AwsEc2ServiceImplTests extends ESTestCase {
         assertThat(configuration.getProxyUsername(), is(expectedProxyUsername));
         assertThat(configuration.getProxyPassword(), is(expectedProxyPassword));
         assertThat(configuration.getSignerOverride(), is(expectedSigner));
+        assertThat(configuration.getSocketTimeout(), is(expectedReadTimeout));
     }
 
     public void testDefaultEndpoint() {

plugins/repository-s3/src/main/java/org/elasticsearch/cloud/aws/AwsS3Service.java

@@ -19,12 +19,14 @@
 
 package org.elasticsearch.cloud.aws;
 
+import com.amazonaws.ClientConfiguration;
 import com.amazonaws.Protocol;
 import com.amazonaws.services.s3.AmazonS3;
 import org.elasticsearch.common.component.LifecycleComponent;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Setting.Property;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.TimeValue;
 
 import java.util.Locale;
 import java.util.function.Function;
@@ -76,6 +78,11 @@ public interface AwsS3Service extends LifecycleComponent {
      */
     Setting<String> REGION_SETTING =
         new Setting<>("cloud.aws.region", "", s -> s.toLowerCase(Locale.ROOT), Property.NodeScope, Property.Shared);
+    /**
+     * cloud.aws.read_timeout: Socket read timeout. Shared with discovery-ec2 plugin
+     */
+    Setting<TimeValue> READ_TIMEOUT = Setting.timeSetting("cloud.aws.read_timeout",
+        TimeValue.timeValueMillis(ClientConfiguration.DEFAULT_SOCKET_TIMEOUT), Property.NodeScope, Property.Shared);
 
     /**
      * Defines specific s3 settings starting with cloud.aws.s3.
@@ -150,6 +157,12 @@ public interface AwsS3Service extends LifecycleComponent {
          * cloud.aws.s3.endpoint: Endpoint. If not set, endpoint will be guessed based on region setting.
          */
         Setting<String> ENDPOINT_SETTING = Setting.simpleString("cloud.aws.s3.endpoint", Property.NodeScope);
+        /**
+         * cloud.aws.s3.read_timeout: Socket read timeout. Defaults to cloud.aws.read_timeout
+         * @see AwsS3Service#READ_TIMEOUT
+         */
+        Setting<TimeValue> READ_TIMEOUT =
+            Setting.timeSetting("cloud.aws.s3.read_timeout", AwsS3Service.READ_TIMEOUT, Property.NodeScope);
     }
 
     AmazonS3 client(Settings repositorySettings, String endpoint, Protocol protocol, String region, Integer maxRetries,

plugins/repository-s3/src/main/java/org/elasticsearch/cloud/aws/InternalAwsS3Service.java

@@ -116,6 +116,8 @@ public class InternalAwsS3Service extends AbstractLifecycleComponent implements
             AwsSigner.configureSigner(awsSigner, clientConfiguration, endpoint);
         }
 
+        clientConfiguration.setSocketTimeout((int) CLOUD_S3.READ_TIMEOUT.get(settings).millis());
+
         return clientConfiguration;
     }
 

plugins/repository-s3/src/main/java/org/elasticsearch/plugin/repository/s3/S3RepositoryPlugin.java

@@ -96,6 +96,7 @@ public class S3RepositoryPlugin extends Plugin implements RepositoryPlugin {
         AwsS3Service.PROXY_PASSWORD_SETTING,
         AwsS3Service.SIGNER_SETTING,
         AwsS3Service.REGION_SETTING,
+        AwsS3Service.READ_TIMEOUT,
 
         // Register S3 specific settings: cloud.aws.s3
         AwsS3Service.CLOUD_S3.KEY_SETTING,
@@ -108,6 +109,7 @@ public class S3RepositoryPlugin extends Plugin implements RepositoryPlugin {
         AwsS3Service.CLOUD_S3.SIGNER_SETTING,
         AwsS3Service.CLOUD_S3.REGION_SETTING,
         AwsS3Service.CLOUD_S3.ENDPOINT_SETTING,
+        AwsS3Service.CLOUD_S3.READ_TIMEOUT,
 
         // Register S3 repositories settings: repositories.s3
         S3Repository.Repositories.KEY_SETTING,

plugins/repository-s3/src/test/java/org/elasticsearch/cloud/aws/AwsS3ServiceImplTests.java

@@ -143,7 +143,8 @@ public class AwsS3ServiceImplTests extends ESTestCase {
 
     public void testAWSDefaultConfiguration() {
         Settings repositorySettings = generateRepositorySettings(null, null, "eu-central", null, null);
-        launchAWSConfigurationTest(Settings.EMPTY, repositorySettings, Protocol.HTTPS, null, -1, null, null, null, 3, false);
+        launchAWSConfigurationTest(Settings.EMPTY, repositorySettings, Protocol.HTTPS, null, -1, null, null, null, 3, false,
+            ClientConfiguration.DEFAULT_SOCKET_TIMEOUT);
     }
 
     public void testAWSConfigurationWithAwsSettings() {
@@ -155,9 +156,10 @@ public class AwsS3ServiceImplTests extends ESTestCase {
             .put(AwsS3Service.PROXY_USERNAME_SETTING.getKey(), "aws_proxy_username")
             .put(AwsS3Service.PROXY_PASSWORD_SETTING.getKey(), "aws_proxy_password")
             .put(AwsS3Service.SIGNER_SETTING.getKey(), "AWS3SignerType")
+            .put(AwsS3Service.READ_TIMEOUT.getKey(), "10s")
             .build();
         launchAWSConfigurationTest(settings, repositorySettings, Protocol.HTTP, "aws_proxy_host", 8080, "aws_proxy_username",
-            "aws_proxy_password", "AWS3SignerType", 3, false);
+            "aws_proxy_password", "AWS3SignerType", 3, false, 10000);
     }
 
     public void testAWSConfigurationWithAwsAndS3Settings() {
@@ -175,9 +177,10 @@ public class AwsS3ServiceImplTests extends ESTestCase {
             .put(AwsS3Service.CLOUD_S3.PROXY_USERNAME_SETTING.getKey(), "s3_proxy_username")
             .put(AwsS3Service.CLOUD_S3.PROXY_PASSWORD_SETTING.getKey(), "s3_proxy_password")
             .put(AwsS3Service.CLOUD_S3.SIGNER_SETTING.getKey(), "NoOpSignerType")
+            .put(AwsS3Service.CLOUD_S3.READ_TIMEOUT.getKey(), "10s")
             .build();
         launchAWSConfigurationTest(settings, repositorySettings, Protocol.HTTPS, "s3_proxy_host", 8081, "s3_proxy_username",
-            "s3_proxy_password", "NoOpSignerType", 3, false);
+            "s3_proxy_password", "NoOpSignerType", 3, false, 10000);
     }
 
     protected void launchAWSConfigurationTest(Settings settings,
@@ -189,7 +192,8 @@ public class AwsS3ServiceImplTests extends ESTestCase {
                                               String expectedProxyPassword,
                                               String expectedSigner,
                                               Integer expectedMaxRetries,
-                                              boolean expectedUseThrottleRetries) {
+                                              boolean expectedUseThrottleRetries,
+                                              int expectedReadTimeout) {
         Protocol protocol = S3Repository.getValue(singleRepositorySettings, settings,
             S3Repository.Repository.PROTOCOL_SETTING, S3Repository.Repositories.PROTOCOL_SETTING);
         Integer maxRetries = S3Repository.getValue(singleRepositorySettings, settings,
@@ -209,6 +213,7 @@ public class AwsS3ServiceImplTests extends ESTestCase {
         assertThat(configuration.getSignerOverride(), is(expectedSigner));
         assertThat(configuration.getMaxErrorRetry(), is(expectedMaxRetries));
         assertThat(configuration.useThrottledRetries(), is(expectedUseThrottleRetries));
+        assertThat(configuration.getSocketTimeout(), is(expectedReadTimeout));
     }
 
     private static Settings generateRepositorySettings(String key, String secret, String region, String endpoint, Integer maxRetries) {