Feature/storage/add cv stress tests

sdk/storage/azure-storage-blob-stress/pom.xml

@@ -57,12 +57,12 @@
     <dependency>
       <groupId>io.opentelemetry.instrumentation</groupId>
       <artifactId>opentelemetry-runtime-telemetry-java8</artifactId>
-      <version>2.24.0-alpha</version> <!-- {x-version-update;io.opentelemetry.instrumentation:opentelemetry-runtime-telemetry-java8;external_dependency} -->
+      <version>2.15.0-alpha</version> <!-- {x-version-update;io.opentelemetry.instrumentation:opentelemetry-runtime-telemetry-java8;external_dependency} -->
     </dependency>
     <dependency>
       <groupId>io.opentelemetry.instrumentation</groupId>
       <artifactId>opentelemetry-logback-appender-1.0</artifactId>
-      <version>2.24.0-alpha</version> <!-- {x-version-update;io.opentelemetry.instrumentation:opentelemetry-logback-appender-1.0;external_dependency} -->
+      <version>2.15.0-alpha</version> <!-- {x-version-update;io.opentelemetry.instrumentation:opentelemetry-logback-appender-1.0;external_dependency} -->
     </dependency>
 
     <dependency>

sdk/storage/azure-storage-blob-stress/scripts/fault-injector.sh

@@ -1,4 +1,4 @@
 #!/bin/sh
 set -ex;
-dotnet dev-certs https --export-path /mnt/outputs/dev-cert.pfx;
+dotnet dev-certs https --export-path /mnt/outputs/dev-cert.crt --format PEM --no-password;
 /root/.dotnet/tools/http-fault-injector;

sdk/storage/azure-storage-blob-stress/scripts/stress-run.sh

@@ -1,4 +1,14 @@
 #!/bin/sh
 set -ex;
 set -exa;
-keytool -import -alias test -file /mnt/outputs/dev-cert.pfx -keystore ${JAVA_HOME}/lib/security/cacerts -noprompt -keypass changeit -storepass changeit;
+attempts=0;
+while [ ! -s /mnt/outputs/dev-cert.crt ]; do
+  attempts=$((attempts + 1));
+  if [ "$attempts" -gt 60 ]; then
+    echo "Timed out waiting for fault injector certificate" >&2;
+    exit 1;
+  fi;
+  sleep 1;
+done;
+keytool -delete -alias HttpFaultInject -keystore "${JAVA_HOME}/lib/security/cacerts" -storepass changeit >/dev/null 2>&1 || true;
+keytool -importcert -trustcacerts -alias HttpFaultInject -file /mnt/outputs/dev-cert.crt -keystore "${JAVA_HOME}/lib/security/cacerts" -noprompt -storepass changeit;

sdk/storage/azure-storage-blob-stress/src/main/java/com/azure/storage/blob/stress/ContentValidationDecoderStressOptions.java

@@ -0,0 +1,26 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.storage.blob.stress;
+
+import com.azure.storage.common.ContentValidationAlgorithm;
+import com.azure.storage.stress.StorageStressOptions;
+import com.beust.jcommander.Parameter;
+
+/**
+ * Options for stress scenarios that enable transactional response content validation on downloads
+ * (CRC64 / structured message). See {@link com.azure.storage.blob.BlobContentValidationDownloadTests}.
+ */
+public class ContentValidationDecoderStressOptions extends StorageStressOptions {
+    /**
+     * Response content validation behavior for download APIs. Use CRC64 or AUTO to exercise content validation.
+     * NONE disables response validation.
+     */
+    @Parameter(names = { "--contentValidationAlgorithm" },
+        description = "CRC64 (default), AUTO, or NONE")
+    private ContentValidationAlgorithm contentValidationAlgorithm = ContentValidationAlgorithm.CRC64;
+
+    public ContentValidationAlgorithm getContentValidationAlgorithm() {
+        return contentValidationAlgorithm;
+    }
+}

sdk/storage/azure-storage-blob-stress/src/main/java/com/azure/storage/blob/stress/ContentValidationDownloadContent.java

@@ -0,0 +1,69 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.storage.blob.stress;
+
+import com.azure.core.http.HttpHeaderName;
+import com.azure.core.util.BinaryData;
+import com.azure.core.util.Context;
+import com.azure.storage.blob.BlobAsyncClient;
+import com.azure.storage.blob.BlobClient;
+import com.azure.storage.blob.options.BlobDownloadContentOptions;
+import com.azure.storage.blob.options.BlobDownloadStreamOptions;
+import com.azure.storage.blob.stress.utils.OriginalContent;
+import reactor.core.publisher.Mono;
+
+/**
+ * Download content with
+ * {@link BlobDownloadContentOptions#setContentValidationAlgorithm} enabled.
+ * Verifies the correctness of the download response content via CRC.
+ */
+public class ContentValidationDownloadContent extends BlobScenarioBase<ContentValidationDecoderStressOptions> {
+    private final OriginalContent originalContent = new OriginalContent();
+    private final BlobClient syncClient;
+    private final BlobAsyncClient asyncClient;
+    private final BlobAsyncClient asyncNoFaultClient;
+
+    public ContentValidationDownloadContent(ContentValidationDecoderStressOptions options) {
+        super(options);
+        String blobName = generateBlobName();
+        this.asyncNoFaultClient = getAsyncContainerClientNoFault().getBlobAsyncClient(blobName);
+        this.syncClient = getSyncContainerClient().getBlobClient(blobName);
+        this.asyncClient = getAsyncContainerClient().getBlobAsyncClient(blobName);
+    }
+
+    @Override
+    protected void runInternal(Context span) {
+        originalContent.checkMatch(
+            syncClient.downloadContentWithResponse(
+                new BlobDownloadContentOptions()
+                    .setContentValidationAlgorithm(options.getContentValidationAlgorithm()),
+                null, span).getValue(),
+            span).block();
+    }
+
+    @Override
+    protected Mono<Void> runInternalAsync(Context span) {
+        // TODO return downloadContent once it stops buffering.
+        return asyncClient.downloadStreamWithResponse(
+            new BlobDownloadStreamOptions()
+                .setContentValidationAlgorithm(options.getContentValidationAlgorithm()))
+            .flatMap(response -> {
+                long contentLength = Long.valueOf(response.getHeaders().getValue(HttpHeaderName.CONTENT_LENGTH));
+                return BinaryData.fromFlux(response.getValue(), contentLength, false);
+            })
+            .flatMap(bd -> originalContent.checkMatch(bd, span));
+    }
+
+    @Override
+    public Mono<Void> setupAsync() {
+        return super.setupAsync()
+            .then(originalContent.setupBlob(asyncNoFaultClient, options.getSize()));
+    }
+
+    @Override
+    public Mono<Void> cleanupAsync() {
+        return asyncNoFaultClient.deleteIfExists()
+            .then(super.cleanupAsync());
+    }
+}

sdk/storage/azure-storage-blob-stress/src/main/java/com/azure/storage/blob/stress/ContentValidationDownloadStream.java

@@ -0,0 +1,66 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.storage.blob.stress;
+
+import com.azure.core.util.Context;
+import com.azure.storage.blob.BlobAsyncClient;
+import com.azure.storage.blob.BlobClient;
+import com.azure.storage.blob.options.BlobDownloadStreamOptions;
+import com.azure.storage.blob.stress.utils.OriginalContent;
+import com.azure.storage.stress.CrcOutputStream;
+import reactor.core.publisher.Mono;
+
+import java.io.IOException;
+
+/**
+ * Streaming blob download with
+ * {@link BlobDownloadStreamOptions#setContentValidationAlgorithm} enabled.
+ * Verifies the correctness of the download response content via CRC.
+ */
+public class ContentValidationDownloadStream extends BlobScenarioBase<ContentValidationDecoderStressOptions> {
+    private final OriginalContent originalContent = new OriginalContent();
+    private final BlobClient syncClient;
+    private final BlobAsyncClient asyncClient;
+    private final BlobAsyncClient asyncNoFaultClient;
+
+    public ContentValidationDownloadStream(ContentValidationDecoderStressOptions options) {
+        super(options);
+        String blobName = generateBlobName();
+        this.asyncNoFaultClient = getAsyncContainerClientNoFault().getBlobAsyncClient(blobName);
+        this.syncClient = getSyncContainerClient().getBlobClient(blobName);
+        this.asyncClient = getAsyncContainerClient().getBlobAsyncClient(blobName);
+    }
+
+    @Override
+    protected void runInternal(Context span) throws IOException {
+        try (CrcOutputStream outputStream = new CrcOutputStream()) {
+            syncClient.downloadStreamWithResponse(outputStream,
+                new BlobDownloadStreamOptions()
+                    .setContentValidationAlgorithm(options.getContentValidationAlgorithm()),
+                null, span);
+            outputStream.close();
+            originalContent.checkMatch(outputStream.getContentInfo(), span).block();
+        }
+    }
+
+    @Override
+    protected Mono<Void> runInternalAsync(Context span) {
+        return asyncClient.downloadStreamWithResponse(
+            new BlobDownloadStreamOptions()
+                .setContentValidationAlgorithm(options.getContentValidationAlgorithm()))
+            .flatMap(response -> originalContent.checkMatch(response.getValue(), span));
+    }
+
+    @Override
+    public Mono<Void> setupAsync() {
+        return super.setupAsync()
+            .then(originalContent.setupBlob(asyncNoFaultClient, options.getSize()));
+    }
+
+    @Override
+    public Mono<Void> cleanupAsync() {
+        return asyncNoFaultClient.deleteIfExists()
+            .then(super.cleanupAsync());
+    }
+}

sdk/storage/azure-storage-blob-stress/src/main/java/com/azure/storage/blob/stress/ContentValidationDownloadToFile.java

@@ -0,0 +1,104 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.storage.blob.stress;
+
+import com.azure.core.util.BinaryData;
+import com.azure.core.util.Context;
+import com.azure.core.util.logging.ClientLogger;
+import com.azure.storage.blob.BlobAsyncClient;
+import com.azure.storage.blob.BlobClient;
+import com.azure.storage.blob.options.BlobDownloadToFileOptions;
+import com.azure.storage.blob.stress.utils.OriginalContent;
+import com.azure.storage.common.ParallelTransferOptions;
+import reactor.core.publisher.Mono;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.UUID;
+
+/**
+ * Download to file with
+ * {@link BlobDownloadToFileOptions#setContentValidationAlgorithm} enabled.
+ * Verifies the correctness of the download response content via CRC.
+ */
+public class ContentValidationDownloadToFile extends BlobScenarioBase<ContentValidationDecoderStressOptions> {
+    private static final ClientLogger LOGGER = new ClientLogger(ContentValidationDownloadToFile.class);
+    private final Path directoryPath;
+    private final OriginalContent originalContent = new OriginalContent();
+    private final BlobClient syncClient;
+    private final BlobAsyncClient asyncClient;
+    private final BlobAsyncClient asyncNoFaultClient;
+    private final ParallelTransferOptions parallelTransferOptions;
+
+    public ContentValidationDownloadToFile(ContentValidationDecoderStressOptions options) {
+        super(options);
+        this.directoryPath = getTempPath("test");
+        String blobName = generateBlobName();
+        this.asyncNoFaultClient = getAsyncContainerClientNoFault().getBlobAsyncClient(blobName);
+        this.syncClient = getSyncContainerClient().getBlobClient(blobName);
+        this.asyncClient = getAsyncContainerClient().getBlobAsyncClient(blobName);
+        this.parallelTransferOptions = new ParallelTransferOptions()
+            .setMaxConcurrency(options.getMaxConcurrency());
+    }
+
+    @Override
+    protected void runInternal(Context span) {
+        Path downloadPath = directoryPath.resolve(UUID.randomUUID() + ".txt");
+        BlobDownloadToFileOptions blobOptions = new BlobDownloadToFileOptions(downloadPath.toString())
+            .setParallelTransferOptions(parallelTransferOptions)
+            .setContentValidationAlgorithm(options.getContentValidationAlgorithm());
+
+        try {
+            syncClient.downloadToFileWithResponse(blobOptions, Duration.ofSeconds(options.getDuration()), span);
+            originalContent.checkMatch(BinaryData.fromFile(downloadPath), span).block();
+        } finally {
+            deleteFile(downloadPath);
+        }
+    }
+
+    @Override
+    protected Mono<Void> runInternalAsync(Context span) {
+        return Mono.using(
+            () -> directoryPath.resolve(UUID.randomUUID() + ".txt"),
+            path -> asyncClient.downloadToFileWithResponse(
+                new BlobDownloadToFileOptions(path.toString())
+                    .setParallelTransferOptions(parallelTransferOptions)
+                    .setContentValidationAlgorithm(options.getContentValidationAlgorithm()))
+                .flatMap(ignored -> originalContent.checkMatch(BinaryData.fromFile(path), span)),
+            ContentValidationDownloadToFile::deleteFile);
+    }
+
+    @Override
+    public Mono<Void> setupAsync() {
+        return super.setupAsync()
+            .then(originalContent.setupBlob(asyncNoFaultClient, options.getSize()));
+    }
+
+    @Override
+    public Mono<Void> cleanupAsync() {
+        return asyncNoFaultClient.deleteIfExists()
+            .then(super.cleanupAsync());
+    }
+
+    private Path getTempPath(String prefix) {
+        try {
+            return Files.createTempDirectory(prefix);
+        } catch (IOException e) {
+            throw LOGGER.logExceptionAsError(new UncheckedIOException(e));
+        }
+    }
+
+    private static void deleteFile(Path path) {
+        try {
+            Files.deleteIfExists(path);
+        } catch (Throwable e) {
+            LOGGER.atError()
+                .addKeyValue("path", path)
+                .log("failed to delete file", e);
+        }
+    }
+}

sdk/storage/azure-storage-blob-stress/src/main/java/com/azure/storage/blob/stress/ContentValidationOpenInputStream.java

@@ -0,0 +1,69 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.storage.blob.stress;
+
+import com.azure.core.util.Context;
+import com.azure.core.util.logging.ClientLogger;
+import com.azure.storage.blob.BlobAsyncClient;
+import com.azure.storage.blob.BlobClient;
+import com.azure.storage.blob.options.BlobInputStreamOptions;
+import com.azure.storage.blob.stress.utils.OriginalContent;
+import com.azure.storage.stress.CrcInputStream;
+import reactor.core.publisher.Mono;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+import static com.azure.core.util.FluxUtil.monoError;
+
+/**
+ * Open input stream with {@link BlobInputStreamOptions#setContentValidationAlgorithm} enabled (sync only).
+ * Verifies the correctness of the download response content via CRC.
+ */
+public class ContentValidationOpenInputStream extends BlobScenarioBase<ContentValidationDecoderStressOptions> {
+    private static final ClientLogger LOGGER = new ClientLogger(ContentValidationOpenInputStream.class);
+    private final OriginalContent originalContent = new OriginalContent();
+    private final BlobClient syncClient;
+    private final BlobAsyncClient asyncNoFaultClient;
+
+    public ContentValidationOpenInputStream(ContentValidationDecoderStressOptions options) {
+        super(options);
+        String blobName = generateBlobName();
+        this.syncClient = getSyncContainerClient().getBlobClient(blobName);
+        this.asyncNoFaultClient = getAsyncContainerClientNoFault().getBlobAsyncClient(blobName);
+    }
+
+    @Override
+    protected void runInternal(Context span) throws IOException {
+        try (InputStream stream = syncClient.openInputStream(
+            new BlobInputStreamOptions()
+                .setContentValidationAlgorithm(options.getContentValidationAlgorithm()),
+            span)) {
+            try (CrcInputStream crcStream = new CrcInputStream(stream)) {
+                byte[] buffer = new byte[8192];
+                while (crcStream.read(buffer) != -1) {
+                    // do nothing
+                }
+                originalContent.checkMatch(crcStream.getContentInfo(), span).block();
+            }
+        }
+    }
+
+    @Override
+    protected Mono<Void> runInternalAsync(Context span) {
+        return monoError(LOGGER, new RuntimeException("openInputStream() does not exist on the async client"));
+    }
+
+    @Override
+    public Mono<Void> setupAsync() {
+        return super.setupAsync()
+            .then(originalContent.setupBlob(asyncNoFaultClient, options.getSize()));
+    }
+
+    @Override
+    public Mono<Void> cleanupAsync() {
+        return asyncNoFaultClient.deleteIfExists()
+            .then(super.cleanupAsync());
+    }
+}