Add timeout settings (default to 5 minutes)

By default, azure does not timeout. This commit adds support for a timeout settings which defaults to 5 minutes.
It's a timeout **per request** not a global timeout for a snapshot request.

It can be defined globally, per account or both. Defaults to `5m`.

```yml
cloud:
    azure:
        storage:
            timeout: 10s
            my_account1:
                account: your_azure_storage_account1
                key: your_azure_storage_key1
                default: true
            my_account2:
                account: your_azure_storage_account2
                key: your_azure_storage_key2
                timeout: 30s
```

In this example, timeout will be 10s for `my_account1` and 30s for `my_account2`.

Backport of #15080 (#15950) in 2.2 branch.

Related to #14277.

docs/plugins/cloud-azure.asciidoc

@@ -543,6 +543,27 @@ cloud:
 
 `my_account1` is the default account which will be used by a repository unless you set an explicit one.
 
+You can set the timeout to use when making any single request. It can be defined globally, per account or both.
+Defaults to `5m`.
+
+[source,yaml]
+----
+cloud:
+    azure:
+        storage:
+            timeout: 10s
+            my_account1:
+                account: your_azure_storage_account1
+                key: your_azure_storage_key1
+                default: true
+            my_account2:
+                account: your_azure_storage_account2
+                key: your_azure_storage_key2
+                timeout: 30s
+----
+
+In this example, timeout will be 10s for `my_account1` and 30s for `my_account2`.
+
 
 The Azure repository supports following settings:
 

plugins/cloud-azure/src/main/java/org/elasticsearch/cloud/azure/storage/AzureStorageService.java

@@ -38,8 +38,14 @@ final class Storage {
         public static final String API_IMPLEMENTATION = "cloud.azure.storage.api.impl";
         public static final String PREFIX = "cloud.azure.storage.";
         @Deprecated
-        public static final String ACCOUNT = "cloud.azure.storage.account";
+        public static final String ACCOUNT_DEPRECATED = "cloud.azure.storage.account";
         @Deprecated
+        public static final String KEY_DEPRECATED = "cloud.azure.storage.key";
+
+        public static final String TIMEOUT = "cloud.azure.storage.timeout";
+
+        public static final String ACCOUNT = "repositories.azure.account";
+        public static final String LOCATION_MODE = "repositories.azure.location_mode";
         public static final String KEY = "cloud.azure.storage.key";
         public static final String CONTAINER = "repositories.azure.container";
         public static final String BASE_PATH = "repositories.azure.base_path";

plugins/cloud-azure/src/main/java/org/elasticsearch/cloud/azure/storage/AzureStorageServiceImpl.java

@@ -47,7 +47,7 @@ public class AzureStorageServiceImpl extends AbstractLifecycleComponent<AzureSto
     final Map<String, AzureStorageSettings> secondariesStorageSettings;
 
     final Map<String, CloudBlobClient> clients;
-    
+
     @Inject
     public AzureStorageServiceImpl(Settings settings) {
         super(settings);
@@ -81,7 +81,7 @@ void createClient(AzureStorageSettings azureStorageSettings) {
             logger.error("can not create azure storage client: {}", e.getMessage());
         }
     }
-    
+
     CloudBlobClient getSelectedClient(String account, LocationMode mode) {
         logger.trace("selecting a client for account [{}], mode [{}]", account, mode.name());
         AzureStorageSettings azureStorageSettings = null;
@@ -115,9 +115,18 @@ CloudBlobClient getSelectedClient(String account, LocationMode mode) {
         // NOTE: for now, just set the location mode in case it is different;
         // only one mode per storage account can be active at a time
         client.getDefaultRequestOptions().setLocationMode(mode);
+
+        // Set timeout option. Defaults to 5mn. See cloud.azure.storage.timeout or cloud.azure.storage.xxx.timeout
+        try {
+            int timeout = (int) azureStorageSettings.getTimeout().getMillis();
+            client.getDefaultRequestOptions().setTimeoutIntervalInMs(timeout);
+        } catch (ClassCastException e) {
+            throw new IllegalArgumentException("Can not convert [" + azureStorageSettings.getTimeout() +
+                "]. It can not be longer than 2,147,483,647ms.");
+        }
         return client;
     }
-    
+
     @Override
     public boolean doesContainerExist(String account, LocationMode mode, String container) {
         try {
@@ -217,9 +226,9 @@ public OutputStream getOutputStream(String account, LocationMode mode, String co
     @Override
     public Map<String, BlobMetaData> listBlobsByPrefix(String account, LocationMode mode, String container, String keyPath, String prefix) throws URISyntaxException, StorageException {
         // NOTE: this should be here: if (prefix == null) prefix = "";
-        // however, this is really inefficient since deleteBlobsByPrefix enumerates everything and 
+        // however, this is really inefficient since deleteBlobsByPrefix enumerates everything and
         // then does a prefix match on the result; it should just call listBlobsByPrefix with the prefix!
-        
+
         logger.debug("listing container [{}], keyPath [{}], prefix [{}]", container, keyPath, prefix);
         MapBuilder<String, BlobMetaData> blobsBuilder = MapBuilder.newMapBuilder();
 

plugins/cloud-azure/src/main/java/org/elasticsearch/cloud/azure/storage/AzureStorageSettings.java

@@ -24,6 +24,9 @@
 import org.elasticsearch.common.logging.ESLogger;
 import org.elasticsearch.common.logging.ESLoggerFactory;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.repositories.RepositorySettings;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -34,11 +37,13 @@ public class AzureStorageSettings {
     private String name;
     private String account;
     private String key;
+    private TimeValue timeout;
 
-    public AzureStorageSettings(String name, String account, String key) {
+    public AzureStorageSettings(String name, String account, String key, TimeValue timeout) {
         this.name = name;
         this.account = account;
         this.key = key;
+        this.timeout = timeout;
     }
 
     public String getName() {
@@ -53,12 +58,17 @@ public String getAccount() {
         return account;
     }
 
+    public TimeValue getTimeout() {
+        return timeout;
+    }
+
     @Override
     public String toString() {
         final StringBuffer sb = new StringBuffer("AzureStorageSettings{");
         sb.append("name='").append(name).append('\'');
         sb.append(", account='").append(account).append('\'');
         sb.append(", key='").append(key).append('\'');
+        sb.append(", timeout=").append(timeout);
         sb.append('}');
         return sb.toString();
     }
@@ -73,12 +83,15 @@ public static Tuple<AzureStorageSettings, Map<String, AzureStorageSettings>> par
         Map<String, AzureStorageSettings> secondaryStorage = new HashMap<>();
 
         // We check for deprecated settings
-        String account = settings.get(Storage.ACCOUNT);
-        String key = settings.get(Storage.KEY);
+        String account = settings.get(Storage.ACCOUNT_DEPRECATED);
+        String key = settings.get(Storage.KEY_DEPRECATED);
+
+        TimeValue globalTimeout = settings.getAsTime(Storage.TIMEOUT, TimeValue.timeValueMinutes(5));
+
         if (account != null) {
             logger.warn("[{}] and [{}] have been deprecated. Use now [{}xxx.account] and [{}xxx.key] where xxx is any name",
-                    Storage.ACCOUNT, Storage.KEY, Storage.PREFIX, Storage.PREFIX);
-            primaryStorage = new AzureStorageSettings(null, account, key);
+                    Storage.ACCOUNT_DEPRECATED, Storage.KEY_DEPRECATED, Storage.PREFIX, Storage.PREFIX);
+            primaryStorage = new AzureStorageSettings(null, account, key, globalTimeout);
         } else {
             Settings storageSettings = settings.getByPrefix(Storage.PREFIX);
             if (storageSettings != null) {
@@ -87,7 +100,8 @@ public static Tuple<AzureStorageSettings, Map<String, AzureStorageSettings>> par
                     if (storage.getValue() instanceof Map) {
                         @SuppressWarnings("unchecked")
                         Map<String, String> map = (Map) storage.getValue();
-                        AzureStorageSettings current = new AzureStorageSettings(storage.getKey(), map.get("account"), map.get("key"));
+                        TimeValue timeout = TimeValue.parseTimeValue(map.get("timeout"), globalTimeout, Storage.PREFIX + storage.getKey() + ".timeout");
+                        AzureStorageSettings current = new AzureStorageSettings(storage.getKey(), map.get("account"), map.get("key"), timeout);
                         String activeStr = map.get("default");
                         boolean activeByDefault = activeStr == null ? false : Boolean.parseBoolean(activeStr);
                         if (activeByDefault) {

plugins/cloud-azure/src/test/java/org/elasticsearch/cloud/azure/storage/AzureStorageServiceTest.java

@@ -37,6 +37,7 @@ public class AzureStorageServiceTest extends ESTestCase {
             .put("cloud.azure.storage.azure2.key", "mykey2")
             .put("cloud.azure.storage.azure3.account", "myaccount3")
             .put("cloud.azure.storage.azure3.key", "mykey3")
+            .put("cloud.azure.storage.azure3.timeout", "30s")
             .build();
 
     public void testGetSelectedClientWithNoPrimaryAndSecondary() {
@@ -116,6 +117,28 @@ public void testGetSelectedClientDefault() {
         assertThat(client.getEndpoint(), is(URI.create("https://azure1")));
     }
 
+    public void testGetSelectedClientGlobalTimeout() {
+        Settings timeoutSettings = Settings.builder()
+                .put(settings)
+                .put("cloud.azure.storage.timeout", "10s")
+                .build();
+
+        AzureStorageServiceImpl azureStorageService = new AzureStorageServiceMock(timeoutSettings);
+        azureStorageService.doStart();
+        CloudBlobClient client1 = azureStorageService.getSelectedClient("azure1", LocationMode.PRIMARY_ONLY);
+        assertThat(client1.getDefaultRequestOptions().getTimeoutIntervalInMs(), is(10 * 1000));
+        CloudBlobClient client3 = azureStorageService.getSelectedClient("azure3", LocationMode.PRIMARY_ONLY);
+        assertThat(client3.getDefaultRequestOptions().getTimeoutIntervalInMs(), is(30 * 1000));
+    }
+
+    public void testGetSelectedClientDefaultTimeout() {
+        AzureStorageServiceImpl azureStorageService = new AzureStorageServiceMock(settings);
+        azureStorageService.doStart();
+        CloudBlobClient client1 = azureStorageService.getSelectedClient("azure1", LocationMode.PRIMARY_ONLY);
+        assertThat(client1.getDefaultRequestOptions().getTimeoutIntervalInMs(), is(5 * 60 * 1000));
+        CloudBlobClient client3 = azureStorageService.getSelectedClient("azure3", LocationMode.PRIMARY_ONLY);
+        assertThat(client3.getDefaultRequestOptions().getTimeoutIntervalInMs(), is(30 * 1000));
+    }
 
     /**
      * This internal class just overload createClient method which is called by AzureStorageServiceImpl.doStart()