Implement CAS support in Azure test fixture (#117104)

Closes ES-5680

Author: nicktindall

modules/repository-azure/src/internalClusterTest/java/org/elasticsearch/repositories/azure/AzureBlobStoreRepositoryTests.java

@@ -9,6 +9,7 @@
 package org.elasticsearch.repositories.azure;
 
 import fixture.azure.AzureHttpHandler;
+import fixture.azure.MockAzureBlobStore;
 
 import com.azure.storage.common.policy.RequestRetryOptions;
 import com.azure.storage.common.policy.RetryPolicyType;
@@ -184,7 +185,12 @@ long getUploadBlockSize() {
     @SuppressForbidden(reason = "this test uses a HttpHandler to emulate an Azure endpoint")
     private static class AzureBlobStoreHttpHandler extends AzureHttpHandler implements BlobStoreHttpHandler {
         AzureBlobStoreHttpHandler(final String account, final String container) {
-            super(account, container, null /* no auth header validation - sometimes it's omitted in these tests (TODO why?) */);
+            super(
+                account,
+                container,
+                null /* no auth header validation - sometimes it's omitted in these tests (TODO why?) */,
+                MockAzureBlobStore.LeaseExpiryPredicate.NEVER_EXPIRE
+            );
         }
     }
 

modules/repository-azure/src/internalClusterTest/java/org/elasticsearch/repositories/azure/AzureStorageCleanupThirdPartyTests.java

@@ -10,6 +10,7 @@
 package org.elasticsearch.repositories.azure;
 
 import fixture.azure.AzureHttpFixture;
+import fixture.azure.MockAzureBlobStore;
 
 import com.azure.core.exception.HttpResponseException;
 import com.azure.storage.blob.BlobContainerClient;
@@ -60,7 +61,8 @@ public class AzureStorageCleanupThirdPartyTests extends AbstractThirdPartyReposi
         System.getProperty("test.azure.container"),
         System.getProperty("test.azure.tenant_id"),
         System.getProperty("test.azure.client_id"),
-        AzureHttpFixture.sharedKeyForAccountPredicate(AZURE_ACCOUNT)
+        AzureHttpFixture.sharedKeyForAccountPredicate(AZURE_ACCOUNT),
+        MockAzureBlobStore.LeaseExpiryPredicate.NEVER_EXPIRE
     );
 
     @Override

modules/repository-azure/src/main/java/org/elasticsearch/repositories/azure/AzureBlobContainer.java

@@ -180,7 +180,7 @@ protected String buildKey(String blobName) {
     }
 
     private boolean skipRegisterOperation(ActionListener<?> listener) {
-        return skipCas(listener) || skipIfNotPrimaryOnlyLocationMode(listener);
+        return skipIfNotPrimaryOnlyLocationMode(listener);
     }
 
     private boolean skipIfNotPrimaryOnlyLocationMode(ActionListener<?> listener) {

modules/repository-azure/src/main/java/org/elasticsearch/repositories/azure/AzureBlobStore.java

@@ -40,6 +40,7 @@
 import com.azure.storage.blob.models.ListBlobsOptions;
 import com.azure.storage.blob.options.BlobParallelUploadOptions;
 import com.azure.storage.blob.options.BlockBlobSimpleUploadOptions;
+import com.azure.storage.blob.specialized.BlobLeaseClient;
 import com.azure.storage.blob.specialized.BlobLeaseClientBuilder;
 import com.azure.storage.blob.specialized.BlockBlobAsyncClient;
 
@@ -1010,7 +1011,7 @@ private static BytesReference innerCompareAndExchangeRegister(
                 }
                 return currentValue;
             } finally {
-                leaseClient.releaseLease();
+                bestEffortRelease(leaseClient);
             }
         } else {
             if (expected.length() == 0) {
@@ -1020,6 +1021,29 @@ private static BytesReference innerCompareAndExchangeRegister(
         }
     }
 
+    /**
+     * Release the lease, ignoring conflicts due to expiry
+     *
+     * @see <a href="https://learn.microsoft.com/en-us/rest/api/storageservices/lease-blob?outcomes-of-lease-operations-on-blobs-by-lease-state">Outcomes of lease operations by lease state</a>
+     * @param leaseClient The client for the lease
+     */
+    private static void bestEffortRelease(BlobLeaseClient leaseClient) {
+        try {
+            leaseClient.releaseLease();
+        } catch (BlobStorageException blobStorageException) {
+            if (blobStorageException.getStatusCode() == RestStatus.CONFLICT.getStatus()) {
+                // This is OK, we tried to release a lease that was expired/re-acquired
+                logger.debug(
+                    "Ignored conflict on release: errorCode={}, message={}",
+                    blobStorageException.getErrorCode(),
+                    blobStorageException.getMessage()
+                );
+            } else {
+                throw blobStorageException;
+            }
+        }
+    }
+
     private static BytesReference downloadRegisterBlob(
         String containerPath,
         String blobKey,

modules/repository-azure/src/test/java/org/elasticsearch/repositories/azure/AzureBlobContainerStatsTests.java

@@ -10,6 +10,7 @@
 package org.elasticsearch.repositories.azure;
 
 import fixture.azure.AzureHttpHandler;
+import fixture.azure.MockAzureBlobStore;
 
 import org.elasticsearch.common.blobstore.OperationPurpose;
 import org.elasticsearch.common.bytes.BytesReference;
@@ -26,7 +27,7 @@ public class AzureBlobContainerStatsTests extends AbstractAzureServerTestCase {
     @SuppressForbidden(reason = "use a http server")
     @Before
     public void configureAzureHandler() {
-        httpServer.createContext("/", new AzureHttpHandler(ACCOUNT, CONTAINER, null));
+        httpServer.createContext("/", new AzureHttpHandler(ACCOUNT, CONTAINER, null, MockAzureBlobStore.LeaseExpiryPredicate.NEVER_EXPIRE));
     }
 
     public void testOperationPurposeIsReflectedInBlobStoreStats() throws IOException {

modules/repository-azure/src/yamlRestTest/java/org/elasticsearch/repositories/azure/RepositoryAzureClientYamlTestSuiteIT.java

@@ -10,6 +10,7 @@
 package org.elasticsearch.repositories.azure;
 
 import fixture.azure.AzureHttpFixture;
+import fixture.azure.MockAzureBlobStore;
 
 import com.carrotsearch.randomizedtesting.annotations.Name;
 import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
@@ -47,7 +48,8 @@ public class RepositoryAzureClientYamlTestSuiteIT extends ESClientYamlSuiteTestC
         AZURE_TEST_CONTAINER,
         AZURE_TEST_TENANT_ID,
         AZURE_TEST_CLIENT_ID,
-        decideAuthHeaderPredicate()
+        decideAuthHeaderPredicate(),
+        MockAzureBlobStore.LeaseExpiryPredicate.NEVER_EXPIRE
     );
 
     private static Predicate<String> decideAuthHeaderPredicate() {

modules/repository-azure/src/yamlRestTest/resources/rest-api-spec/test/repository_azure/20_repository.yml

@@ -193,6 +193,20 @@ setup:
             container: zHHkfSqlbnBsbpSgvCYtxrEfFLqghXtyPvvvKPNBnRCicNHQLE
             client: integration_test
 
+---
+"Register a read-only repository with a non existing container":
+
+  - do:
+      catch: /repository_verification_exception/
+      snapshot.create_repository:
+        repository: repository
+        body:
+          type: azure
+          settings:
+            container: zHHkfSqlbnBsbpSgvCYtxrEfFLqghXtyPvvvKPNBnRCicNHQLE
+            client: integration_test
+            readonly: true
+
 ---
 "Register a repository with a non existing client":
 

test/fixtures/azure-fixture/src/main/java/fixture/azure/AzureHttpFixture.java

@@ -45,6 +45,7 @@ public class AzureHttpFixture extends ExternalResource {
     private final String clientId;
     private final String tenantId;
     private final Predicate<String> authHeaderPredicate;
+    private final MockAzureBlobStore.LeaseExpiryPredicate leaseExpiryPredicate;
 
     private HttpServer server;
     private HttpServer metadataServer;
@@ -116,7 +117,8 @@ public AzureHttpFixture(
         String container,
         @Nullable String rawTenantId,
         @Nullable String rawClientId,
-        Predicate<String> authHeaderPredicate
+        Predicate<String> authHeaderPredicate,
+        MockAzureBlobStore.LeaseExpiryPredicate leaseExpiryPredicate
     ) {
         final var tenantId = Strings.hasText(rawTenantId) ? rawTenantId : null;
         final var clientId = Strings.hasText(rawClientId) ? rawClientId : null;
@@ -135,6 +137,7 @@ public AzureHttpFixture(
         this.tenantId = tenantId;
         this.clientId = clientId;
         this.authHeaderPredicate = authHeaderPredicate;
+        this.leaseExpiryPredicate = leaseExpiryPredicate;
     }
 
     private String scheme() {
@@ -193,7 +196,10 @@ protected void before() {
                 }
                 case HTTP -> {
                     server = HttpServer.create(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0), 0);
-                    server.createContext("/" + account, new AzureHttpHandler(account, container, actualAuthHeaderPredicate));
+                    server.createContext(
+                        "/" + account,
+                        new AzureHttpHandler(account, container, actualAuthHeaderPredicate, leaseExpiryPredicate)
+                    );
                     server.start();
 
                     oauthTokenServiceServer = HttpServer.create(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0), 0);
@@ -222,7 +228,10 @@ protected void before() {
                         final var httpsServer = HttpsServer.create(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0), 0);
                         this.server = httpsServer;
                         httpsServer.setHttpsConfigurator(new HttpsConfigurator(sslContext));
-                        httpsServer.createContext("/" + account, new AzureHttpHandler(account, container, actualAuthHeaderPredicate));
+                        httpsServer.createContext(
+                            "/" + account,
+                            new AzureHttpHandler(account, container, actualAuthHeaderPredicate, leaseExpiryPredicate)
+                        );
                         httpsServer.start();
                     }
                     {