Add support for secondary azure storage account

Follow up for #13228.

This commit adds support for a secondary storage account:

```yml
cloud:
    azure:
        storage:
            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
```

When creating a repository, you can choose which azure account you want to use for it:

```sh
curl -XPUT localhost:9200/_snapshot/my_backup1?pretty -d '{
  "type": "azure"
}'

curl -XPUT localhost:9200/_snapshot/my_backup2?pretty -d '{
  "type": "azure",
  "settings": {
    "account" : "my_account2",
    "location_mode": "secondary_only"
  }
}'
```

`location_mode` supports `primary_only` or `secondary_only`. Defaults to `primary_only`. Note that if you set it
to `secondary_only`, it will force `read_only` to true.

(cherry picked from commit 79a4d9c)

# Conflicts:
#	docs/plugins/repository-azure.asciidoc
#	plugins/cloud-azure/src/main/java/org/elasticsearch/plugin/cloud/azure/CloudAzurePlugin.java
#	plugins/cloud-azure/src/test/java/org/elasticsearch/repositories/azure/AzureSnapshotRestoreTests.java
#	plugins/repository-azure/src/main/java/org/elasticsearch/cloud/azure/AzureRepositoryModule.java

docs/plugins/cloud-azure.asciidoc

@@ -520,22 +520,37 @@ To enable Azure repositories, you have first to set your azure storage settings
 cloud:
     azure:
         storage:
-            account: your_azure_storage_account
-            key: your_azure_storage_key
+            my_account:
+                account: your_azure_storage_account
+                key: your_azure_storage_key
 ----
 
-For information, in previous version of the azure plugin, settings were:
+Note that you can also define more than one account:
 
 [source,yaml]
 ----
 cloud:
     azure:
-        storage_account: your_azure_storage_account
-        storage_key: your_azure_storage_key
+        storage:
+            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
 ----
 
+`my_account1` is the default account which will be used by a repository unless you set an explicit one.
+
+
 The Azure repository supports following settings:
 
+`account`::
+
+    Azure account settings to use. Defaults to the only one if you set a single
+    account or to the one marked as `default` if you have more than one.
+
 `container`::
 
     Container name. Defaults to `elasticsearch-snapshots`
@@ -559,7 +574,12 @@ The Azure repository supports following settings:
 
 `read_only`::
 
-    Makes repository read-only. Defaults to `false`.
+    Makes repository read-only. coming[2.1.0]  Defaults to `false`.
+
+`location_mode`::
+
+    `primary_only` or `secondary_only`. Defaults to `primary_only`. Note that if you set it
+    to `secondary_only`, it will force `read_only` to true.
 
 Some examples, using scripts:
 
@@ -582,14 +602,32 @@ PUT _snapshot/my_backup2
         "compress": true
     }
 }
+
+
+# With two accounts defined in elasticsearch.yml (my_account1 and my_account2)
+PUT _snapshot/my_backup3
+{
+    "type": "azure",
+    "settings": {
+        "account": "my_account1"
+    }
+}
+PUT _snapshot/my_backup4
+{
+    "type": "azure",
+    "settings": {
+        "account": "my_account2",
+        "location_mode": "primary_only"
+    }
+}
 ----
 // AUTOSENSE
 
 Example using Java:
 
 [source,java]
 ----
-client.admin().cluster().preparePutRepository("my_backup3")
+client.admin().cluster().preparePutRepository("my_backup_java1")
     .setType("azure").setSettings(Settings.settingsBuilder()
         .put(Storage.CONTAINER, "backup_container")
         .put(Storage.CHUNK_SIZE, new ByteSizeValue(32, ByteSizeUnit.MB))

plugins/cloud-azure/src/main/java/org/elasticsearch/cloud/azure/AzureModule.java

@@ -164,5 +164,4 @@ public static boolean isPropertyMissing(Settings settings, String name) throws E
         }
         return false;
     }
-
 }

plugins/cloud-azure/src/main/java/org/elasticsearch/cloud/azure/blobstore/AzureBlobContainer.java

@@ -61,7 +61,7 @@ public AzureBlobContainer(String repositoryName, BlobPath path, AzureBlobStore b
     @Override
     public boolean blobExists(String blobName) {
         try {
-            return blobStore.client().blobExists(blobStore.container(), buildKey(blobName));
+            return blobStore.blobExists(blobStore.container(), buildKey(blobName));
         } catch (URISyntaxException | StorageException e) {
             logger.warn("can not access [{}] in container {{}}: {}", blobName, blobStore.container(), e.getMessage());
         }
@@ -71,7 +71,7 @@ public boolean blobExists(String blobName) {
     @Override
     public InputStream openInput(String blobName) throws IOException {
         try {
-            return blobStore.client().getInputStream(blobStore.container(), buildKey(blobName));
+            return blobStore.getInputStream(blobStore.container(), buildKey(blobName));
         } catch (StorageException e) {
             if (e.getHttpStatusCode() == HttpURLConnection.HTTP_NOT_FOUND) {
                 throw new FileNotFoundException(e.getMessage());
@@ -85,7 +85,7 @@ public InputStream openInput(String blobName) throws IOException {
     @Override
     public OutputStream createOutput(String blobName) throws IOException {
         try {
-            return new AzureOutputStream(blobStore.client().getOutputStream(blobStore.container(), buildKey(blobName)));
+            return new AzureOutputStream(blobStore.getOutputStream(blobStore.container(), buildKey(blobName)));
         } catch (StorageException e) {
             if (e.getHttpStatusCode() == HttpURLConnection.HTTP_NOT_FOUND) {
                 throw new FileNotFoundException(e.getMessage());
@@ -101,7 +101,7 @@ public OutputStream createOutput(String blobName) throws IOException {
     @Override
     public void deleteBlob(String blobName) throws IOException {
         try {
-            blobStore.client().deleteBlob(blobStore.container(), buildKey(blobName));
+            blobStore.deleteBlob(blobStore.container(), buildKey(blobName));
         } catch (URISyntaxException | StorageException e) {
             logger.warn("can not access [{}] in container {{}}: {}", blobName, blobStore.container(), e.getMessage());
             throw new IOException(e);
@@ -112,7 +112,7 @@ public void deleteBlob(String blobName) throws IOException {
     public Map<String, BlobMetaData> listBlobsByPrefix(@Nullable String prefix) throws IOException {
 
         try {
-            return blobStore.client().listBlobsByPrefix(blobStore.container(), keyPath, prefix);
+            return blobStore.listBlobsByPrefix(blobStore.container(), keyPath, prefix);
         } catch (URISyntaxException | StorageException e) {
             logger.warn("can not access [{}] in container {{}}: {}", prefix, blobStore.container(), e.getMessage());
             throw new IOException(e);
@@ -127,7 +127,7 @@ public void move(String sourceBlobName, String targetBlobName) throws IOExceptio
 
             logger.debug("moving blob [{}] to [{}] in container {{}}", source, target, blobStore.container());
 
-            blobStore.client().moveBlob(blobStore.container(), source, target);
+            blobStore.moveBlob(blobStore.container(), source, target);
         } catch (URISyntaxException e) {
             logger.warn("can not move blob [{}] to [{}] in container {{}}: {}", sourceBlobName, targetBlobName, blobStore.container(), e.getMessage());
             throw new IOException(e);

plugins/cloud-azure/src/main/java/org/elasticsearch/cloud/azure/blobstore/AzureBlobStore.java

@@ -19,9 +19,11 @@
 
 package org.elasticsearch.cloud.azure.blobstore;
 
+import com.microsoft.azure.storage.LocationMode;
 import com.microsoft.azure.storage.StorageException;
 import org.elasticsearch.cloud.azure.storage.AzureStorageService;
 import org.elasticsearch.common.blobstore.BlobContainer;
+import org.elasticsearch.common.blobstore.BlobMetaData;
 import org.elasticsearch.common.blobstore.BlobPath;
 import org.elasticsearch.common.blobstore.BlobStore;
 import org.elasticsearch.common.component.AbstractComponent;
@@ -30,39 +32,49 @@
 import org.elasticsearch.repositories.RepositoryName;
 import org.elasticsearch.repositories.RepositorySettings;
 
+import java.io.InputStream;
+import java.io.OutputStream;
 import java.net.URISyntaxException;
+import java.util.Locale;
+import java.util.Map;
 
 import static org.elasticsearch.cloud.azure.storage.AzureStorageService.Storage.CONTAINER;
 import static org.elasticsearch.repositories.azure.AzureRepository.CONTAINER_DEFAULT;
+import static org.elasticsearch.repositories.azure.AzureRepository.Repository;
 
-/**
- *
- */
 public class AzureBlobStore extends AbstractComponent implements BlobStore {
 
     private final AzureStorageService client;
 
+    private final String accountName;
+    private final LocationMode locMode;
     private final String container;
     private final String repositoryName;
 
     @Inject
     public AzureBlobStore(RepositoryName name, Settings settings, RepositorySettings repositorySettings,
                           AzureStorageService client) throws URISyntaxException, StorageException {
         super(settings);
-        this.client = client;
+        this.client = client.start();
         this.container = repositorySettings.settings().get("container", settings.get(CONTAINER, CONTAINER_DEFAULT));
         this.repositoryName = name.getName();
+
+        // NOTE: null account means to use the first one specified in config
+        this.accountName = repositorySettings.settings().get(Repository.ACCOUNT, null);
+
+        String modeStr = repositorySettings.settings().get(Repository.LOCATION_MODE, null);
+        if (modeStr == null) {
+            this.locMode = LocationMode.PRIMARY_ONLY;
+        } else {
+            this.locMode = LocationMode.valueOf(modeStr.toUpperCase(Locale.ROOT));
+        }
     }
 
     @Override
     public String toString() {
         return container;
     }
 
-    public AzureStorageService client() {
-        return client;
-    }
-
     public String container() {
         return container;
     }
@@ -80,7 +92,7 @@ public void delete(BlobPath path) {
         }
 
         try {
-            client.deleteFiles(container, keyPath);
+            this.client.deleteFiles(this.accountName, this.locMode, container, keyPath);
         } catch (URISyntaxException | StorageException e) {
             logger.warn("can not remove [{}] in container {{}}: {}", keyPath, container, e.getMessage());
         }
@@ -89,4 +101,54 @@ public void delete(BlobPath path) {
     @Override
     public void close() {
     }
+
+    public boolean doesContainerExist(String container)
+    {
+        return this.client.doesContainerExist(this.accountName, this.locMode, container);
+    }
+
+    public void removeContainer(String container) throws URISyntaxException, StorageException
+    {
+        this.client.removeContainer(this.accountName, this.locMode, container);
+    }
+
+    public void createContainer(String container) throws URISyntaxException, StorageException
+    {
+        this.client.createContainer(this.accountName, this.locMode, container);
+    }
+
+    public void deleteFiles(String container, String path) throws URISyntaxException, StorageException
+    {
+        this.client.deleteFiles(this.accountName, this.locMode, container, path);
+    }
+
+    public boolean blobExists(String container, String blob) throws URISyntaxException, StorageException
+    {
+        return this.client.blobExists(this.accountName, this.locMode, container, blob);
+    }
+
+    public void deleteBlob(String container, String blob) throws URISyntaxException, StorageException
+    {
+        this.client.deleteBlob(this.accountName, this.locMode, container, blob);
+    }
+
+    public InputStream getInputStream(String container, String blob) throws URISyntaxException, StorageException
+    {
+        return this.client.getInputStream(this.accountName, this.locMode, container, blob);
+    }
+
+    public OutputStream getOutputStream(String container, String blob) throws URISyntaxException, StorageException
+    {
+        return this.client.getOutputStream(this.accountName, this.locMode, container, blob);
+    }
+
+    public Map<String,BlobMetaData> listBlobsByPrefix(String container, String keyPath, String prefix) throws URISyntaxException, StorageException
+    {
+        return this.client.listBlobsByPrefix(this.accountName, this.locMode, container, keyPath, prefix);
+    }
+
+    public void moveBlob(String container, String sourceBlob, String targetBlob) throws URISyntaxException, StorageException
+    {
+        this.client.moveBlob(this.accountName, this.locMode, container, sourceBlob, targetBlob);
+    }
 }

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

@@ -20,6 +20,7 @@
 package org.elasticsearch.cloud.azure.storage;
 
 import com.microsoft.azure.storage.StorageException;
+import com.microsoft.azure.storage.LocationMode;
 import org.elasticsearch.common.blobstore.BlobMetaData;
 
 import java.io.InputStream;
@@ -32,33 +33,39 @@
  * @see AzureStorageServiceImpl for Azure REST API implementation
  */
 public interface AzureStorageService {
-    static public final class Storage {
+
+    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";
+        @Deprecated
         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";
         public static final String CHUNK_SIZE = "repositories.azure.chunk_size";
         public static final String COMPRESS = "repositories.azure.compress";
     }
 
-    boolean doesContainerExist(String container);
+    boolean doesContainerExist(String account, LocationMode mode, String container);
+
+    void removeContainer(String account, LocationMode mode, String container) throws URISyntaxException, StorageException;
 
-    void removeContainer(String container) throws URISyntaxException, StorageException;
+    void createContainer(String account, LocationMode mode, String container) throws URISyntaxException, StorageException;
 
-    void createContainer(String container) throws URISyntaxException, StorageException;
+    void deleteFiles(String account, LocationMode mode, String container, String path) throws URISyntaxException, StorageException;
 
-    void deleteFiles(String container, String path) throws URISyntaxException, StorageException;
+    boolean blobExists(String account, LocationMode mode, String container, String blob) throws URISyntaxException, StorageException;
 
-    boolean blobExists(String container, String blob) throws URISyntaxException, StorageException;
+    void deleteBlob(String account, LocationMode mode, String container, String blob) throws URISyntaxException, StorageException;
 
-    void deleteBlob(String container, String blob) throws URISyntaxException, StorageException;
+    InputStream getInputStream(String account, LocationMode mode, String container, String blob) throws URISyntaxException, StorageException;
 
-    InputStream getInputStream(String container, String blob) throws URISyntaxException, StorageException;
+    OutputStream getOutputStream(String account, LocationMode mode, String container, String blob) throws URISyntaxException, StorageException;
 
-    OutputStream getOutputStream(String container, String blob) throws URISyntaxException, StorageException;
+    Map<String,BlobMetaData> listBlobsByPrefix(String account, LocationMode mode, String container, String keyPath, String prefix) throws URISyntaxException, StorageException;
 
-    Map<String,BlobMetaData> listBlobsByPrefix(String container, String keyPath, String prefix) throws URISyntaxException, StorageException;
+    void moveBlob(String account, LocationMode mode, String container, String sourceBlob, String targetBlob) throws URISyntaxException, StorageException;
 
-    void moveBlob(String container, String sourceBlob, String targetBlob) throws URISyntaxException, StorageException;
+    AzureStorageService start();
 }