Use common retry logic for GCS

modules/repository-gcs/src/internalClusterTest/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageBlobStoreRepositoryTests.java

@@ -13,7 +13,6 @@
 import fixture.gcs.GoogleCloudStorageHttpHandler;
 import fixture.gcs.TestUtils;
 
-import com.google.api.gax.retrying.RetrySettings;
 import com.google.cloud.http.HttpTransportOptions;
 import com.google.cloud.storage.StorageOptions;
 import com.google.cloud.storage.StorageRetryStrategy;
@@ -61,8 +60,10 @@
 
 import static org.elasticsearch.common.io.Streams.readFully;
 import static org.elasticsearch.repositories.blobstore.BlobStoreTestUtil.randomPurpose;
+import static org.elasticsearch.repositories.blobstore.BlobStoreTestUtil.randomRetryingPurpose;
 import static org.elasticsearch.repositories.gcs.GoogleCloudStorageClientSettings.CREDENTIALS_FILE_SETTING;
 import static org.elasticsearch.repositories.gcs.GoogleCloudStorageClientSettings.ENDPOINT_SETTING;
+import static org.elasticsearch.repositories.gcs.GoogleCloudStorageClientSettings.MAX_RETRIES_SETTING;
 import static org.elasticsearch.repositories.gcs.GoogleCloudStorageClientSettings.TOKEN_URI_SETTING;
 import static org.elasticsearch.repositories.gcs.GoogleCloudStorageRepository.BASE_PATH;
 import static org.elasticsearch.repositories.gcs.GoogleCloudStorageRepository.BUCKET;
@@ -118,6 +119,7 @@ protected Settings nodeSettings(int nodeOrdinal, Settings otherSettings) {
         settings.put(super.nodeSettings(nodeOrdinal, otherSettings));
         settings.put(ENDPOINT_SETTING.getConcreteSettingForNamespace("test").getKey(), httpServerUrl());
         settings.put(TOKEN_URI_SETTING.getConcreteSettingForNamespace("test").getKey(), httpServerUrl() + "/token");
+        settings.put(MAX_RETRIES_SETTING.getConcreteSettingForNamespace("test").getKey(), 6);
 
         final MockSecureSettings secureSettings = new MockSecureSettings();
         final byte[] serviceAccount = TestUtils.createServiceAccount(random());
@@ -195,7 +197,7 @@ public void testWriteReadLarge() throws IOException {
                 random().nextBytes(data);
                 writeBlob(container, "foobar", new BytesArray(data), false);
             }
-            try (InputStream stream = container.readBlob(randomPurpose(), "foobar")) {
+            try (InputStream stream = container.readBlob(randomRetryingPurpose(), "foobar")) {
                 BytesRefBuilder target = new BytesRefBuilder();
                 while (target.length() < data.length) {
                     byte[] buffer = new byte[scaledRandomIntBetween(1, data.length - target.length())];
@@ -218,7 +220,7 @@ public void testWriteFileMultipleOfChunkSize() throws IOException {
             byte[] initialValue = randomByteArrayOfLength(uploadSize);
             container.writeBlob(randomPurpose(), key, new BytesArray(initialValue), true);
 
-            BytesReference reference = readFully(container.readBlob(randomPurpose(), key));
+            BytesReference reference = readFully(container.readBlob(randomRetryingPurpose(), key));
             assertEquals(new BytesArray(initialValue), reference);
 
             container.deleteBlobsIgnoringIfNotExists(randomPurpose(), Iterators.single(key));
@@ -242,24 +244,18 @@ protected GoogleCloudStorageService createStorageService(ClusterService clusterS
                 @Override
                 StorageOptions createStorageOptions(
                     final GoogleCloudStorageClientSettings gcsClientSettings,
-                    final HttpTransportOptions httpTransportOptions
+                    final HttpTransportOptions httpTransportOptions,
+                    final RetryBehaviour retryBehaviour
                 ) {
-                    StorageOptions options = super.createStorageOptions(gcsClientSettings, httpTransportOptions);
+                    StorageOptions options = super.createStorageOptions(gcsClientSettings, httpTransportOptions, retryBehaviour);
                     return options.toBuilder()
                         .setStorageRetryStrategy(StorageRetryStrategy.getLegacyStorageRetryStrategy())
-                        .setHost(options.getHost())
-                        .setCredentials(options.getCredentials())
                         .setRetrySettings(
-                            RetrySettings.newBuilder()
-                                .setTotalTimeout(options.getRetrySettings().getTotalTimeout())
+                            options.getRetrySettings()
+                                .toBuilder()
                                 .setInitialRetryDelay(Duration.ofMillis(10L))
-                                .setRetryDelayMultiplier(options.getRetrySettings().getRetryDelayMultiplier())
                                 .setMaxRetryDelay(Duration.ofSeconds(1L))
-                                .setMaxAttempts(0)
                                 .setJittered(false)
-                                .setInitialRpcTimeout(options.getRetrySettings().getInitialRpcTimeout())
-                                .setRpcTimeoutMultiplier(options.getRetrySettings().getRpcTimeoutMultiplier())
-                                .setMaxRpcTimeout(options.getRetrySettings().getMaxRpcTimeout())
                                 .build()
                         )
                         .build();

modules/repository-gcs/src/internalClusterTest/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageThirdPartyTests.java

@@ -34,6 +34,7 @@
 
 import static org.elasticsearch.common.io.Streams.readFully;
 import static org.elasticsearch.repositories.blobstore.BlobStoreTestUtil.randomPurpose;
+import static org.elasticsearch.repositories.blobstore.BlobStoreTestUtil.randomRetryingPurpose;
 import static org.hamcrest.Matchers.blankOrNullString;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
@@ -115,7 +116,7 @@ public void testResumeAfterUpdate() {
         executeOnBlobStore(repo, container -> {
             container.writeBlob(randomPurpose(), blobKey, new BytesArray(initialValue), true);
 
-            try (InputStream inputStream = container.readBlob(randomPurpose(), blobKey)) {
+            try (InputStream inputStream = container.readBlob(randomRetryingPurpose(), blobKey)) {
                 // Trigger the first request for the blob, partially read it
                 int read = inputStream.read();
                 assert read != -1;

modules/repository-gcs/src/main/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageBlobStore.java

@@ -139,8 +139,32 @@ class GoogleCloudStorageBlobStore implements BlobStore {
         this.casBackoffPolicy = casBackoffPolicy;
     }
 
-    private MeteredStorage client() throws IOException {
-        return storageService.client(projectId, clientName, repositoryName, statsCollector);
+    /**
+     * Get a client that will retry according to its configured settings
+     *
+     * @return A client
+     */
+    MeteredStorage client() throws IOException {
+        return storageService.client(
+            projectId,
+            clientName,
+            repositoryName,
+            statsCollector,
+            GoogleCloudStorageService.RetryBehaviour.ClientConfigured
+        );
+    }
+
+    /**
+     * Get a client that will not retry on failure
+     *
+     * @return A client with max retries configured to zero
+     */
+    MeteredStorage clientNoRetries() throws IOException {
+        return storageService.client(projectId, clientName, repositoryName, statsCollector, GoogleCloudStorageService.RetryBehaviour.None);
+    }
+
+    int getMaxRetries() {
+        return storageService.clientSettings(projectId, clientName).getMaxRetries();
     }
 
     @Override
@@ -229,7 +253,7 @@ boolean blobExists(OperationPurpose purpose, String blobName) throws IOException
      * @return the InputStream used to read the blob's content
      */
     InputStream readBlob(OperationPurpose purpose, String blobName) throws IOException {
-        return new GoogleCloudStorageRetryingInputStream(purpose, client(), BlobId.of(bucketName, blobName));
+        return new GoogleCloudStorageRetryingInputStream(this, purpose, BlobId.of(bucketName, blobName));
     }
 
     /**
@@ -252,8 +276,8 @@ InputStream readBlob(OperationPurpose purpose, String blobName, long position, l
             return new ByteArrayInputStream(new byte[0]);
         } else {
             return new GoogleCloudStorageRetryingInputStream(
+                this,
                 purpose,
-                client(),
                 BlobId.of(bucketName, blobName),
                 position,
                 Math.addExact(position, length - 1)

modules/repository-gcs/src/main/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageClientSettings.java

@@ -120,6 +120,17 @@ public class GoogleCloudStorageClientSettings {
         () -> PROXY_HOST_SETTING
     );
 
+    /**
+     * The maximum number of retries to use when a GCS request fails.
+     * <p>
+     * Default to 5 to match {@link com.google.cloud.ServiceOptions#getDefaultRetrySettings()}
+     */
+    static final Setting.AffixSetting<Integer> MAX_RETRIES_SETTING = Setting.affixKeySetting(
+        PREFIX,
+        "max_retries",
+        (key) -> Setting.intSetting(key, 5, 0, Setting.Property.NodeScope)
+    );
+
     /** The credentials used by the client to connect to the Storage endpoint. */
     private final ServiceAccountCredentials credential;
 
@@ -144,6 +155,8 @@ public class GoogleCloudStorageClientSettings {
     @Nullable
     private final Proxy proxy;
 
+    private final int maxRetries;
+
     GoogleCloudStorageClientSettings(
         final ServiceAccountCredentials credential,
         final String endpoint,
@@ -152,7 +165,8 @@ public class GoogleCloudStorageClientSettings {
         final TimeValue readTimeout,
         final String applicationName,
         final URI tokenUri,
-        final Proxy proxy
+        final Proxy proxy,
+        final int maxRetries
     ) {
         this.credential = credential;
         this.endpoint = endpoint;
@@ -162,6 +176,7 @@ public class GoogleCloudStorageClientSettings {
         this.applicationName = applicationName;
         this.tokenUri = tokenUri;
         this.proxy = proxy;
+        this.maxRetries = maxRetries;
     }
 
     public ServiceAccountCredentials getCredential() {
@@ -197,6 +212,10 @@ public Proxy getProxy() {
         return proxy;
     }
 
+    public int getMaxRetries() {
+        return maxRetries;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (o == null || getClass() != o.getClass()) return false;
@@ -249,7 +268,8 @@ static GoogleCloudStorageClientSettings getClientSettings(final Settings setting
             getConfigValue(settings, clientName, READ_TIMEOUT_SETTING),
             getConfigValue(settings, clientName, APPLICATION_NAME_SETTING),
             getConfigValue(settings, clientName, TOKEN_URI_SETTING),
-            proxy
+            proxy,
+            getConfigValue(settings, clientName, MAX_RETRIES_SETTING)
         );
     }
 

modules/repository-gcs/src/main/java/org/elasticsearch/repositories/gcs/GoogleCloudStoragePlugin.java

@@ -93,7 +93,8 @@ public List<Setting<?>> getSettings() {
             GoogleCloudStorageClientSettings.TOKEN_URI_SETTING,
             GoogleCloudStorageClientSettings.PROXY_TYPE_SETTING,
             GoogleCloudStorageClientSettings.PROXY_HOST_SETTING,
-            GoogleCloudStorageClientSettings.PROXY_PORT_SETTING
+            GoogleCloudStorageClientSettings.PROXY_PORT_SETTING,
+            GoogleCloudStorageClientSettings.MAX_RETRIES_SETTING
         );
     }