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`.

Closes #14277.

docs/plugins/repository-azure.asciidoc

@@ -64,6 +64,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`.
+
 [[repository-azure-repository-settings]]
 ===== Repository settings
 

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

@@ -41,6 +41,8 @@ public interface AzureStorageService {
         @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 CONTAINER = "repositories.azure.container";

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

@@ -119,6 +119,14 @@ public class AzureStorageServiceImpl extends AbstractLifecycleComponent<AzureSto
         // 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 cast [" + azureStorageSettings.getTimeout() + "] to int.");
+        }
         return client;
     }
 

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

@@ -25,6 +25,7 @@ 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;
@@ -36,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() {
@@ -55,12 +58,17 @@ public class AzureStorageSettings {
         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();
     }
@@ -77,10 +85,13 @@ public class AzureStorageSettings {
         // We check for deprecated settings
         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_DEPRECATED, Storage.KEY_DEPRECATED, Storage.PREFIX, Storage.PREFIX);
-            primaryStorage = new AzureStorageSettings(null, account, key);
+            primaryStorage = new AzureStorageSettings(null, account, key, globalTimeout);
         } else {
             Settings storageSettings = settings.getByPrefix(Storage.PREFIX);
             if (storageSettings != null) {
@@ -89,7 +100,8 @@ public class AzureStorageSettings {
                     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);
                         boolean activeByDefault = Boolean.parseBoolean(map.getOrDefault("default", "false"));
                         if (activeByDefault) {
                             if (primaryStorage == null) {

plugins/repository-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() {
@@ -89,6 +90,28 @@ public class AzureStorageServiceTest extends ESTestCase {
         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()