Merge branch 'main' into fix-validation-test

Author: smalyshev

docs/changelog/128218.yaml

@@ -0,0 +1,5 @@
+pr: 128218
+summary: Improve exception for trained model deployment scale up timeout
+area: Machine Learning
+type: enhancement
+issues: []

docs/release-notes/index.md

@@ -409,6 +409,8 @@ Machine Learning:
 * Support mTLS for the Elastic Inference Service integration inside the inference API [#119679](https://github.com/elastic/elasticsearch/pull/119679)
 * [Inference API] Add node-local rate limiting for the inference API [#120400](https://github.com/elastic/elasticsearch/pull/120400)
 * [Inference API] fix spell words: covertToString to convertToString [#119922](https://github.com/elastic/elasticsearch/pull/119922)
+* Update Linux build images to Rocky Linux 8 with gcc 13.3 [#2773](https://github.com/elastic/ml-cpp/pull/2773)
+
 
 Mapping:
 * Add Optional Source Filtering to Source Loaders [#113827](https://github.com/elastic/elasticsearch/pull/113827)

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

@@ -39,13 +39,21 @@
 import org.elasticsearch.rest.RestStatus;
 import org.junit.ClassRule;
 
-import java.io.ByteArrayInputStream;
+import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
 import java.net.HttpURLConnection;
+import java.nio.channels.Channels;
+import java.nio.file.Files;
 import java.util.Collection;
+import java.util.zip.CRC32;
+import java.util.zip.CheckedInputStream;
+import java.util.zip.CheckedOutputStream;
 
+import static org.elasticsearch.common.io.Streams.limitStream;
 import static org.elasticsearch.repositories.blobstore.BlobStoreTestUtil.randomPurpose;
 import static org.hamcrest.Matchers.blankOrNullString;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.lessThan;
 import static org.hamcrest.Matchers.not;
 
 /**
@@ -58,6 +66,27 @@ public class AzureStorageCleanupThirdPartyTests extends AbstractThirdPartyReposi
 
     private static final String AZURE_ACCOUNT = System.getProperty("test.azure.account");
 
+    /**
+     * AzureRepositoryPlugin that sets a low value for getUploadBlockSize()
+     */
+    public static class TestAzureRepositoryPlugin extends AzureRepositoryPlugin {
+
+        public TestAzureRepositoryPlugin(Settings settings) {
+            super(settings);
+        }
+
+        @Override
+        AzureStorageService createAzureStorageService(Settings settings, AzureClientProvider azureClientProvider) {
+            final long blockSize = ByteSizeValue.ofKb(64L).getBytes() * randomIntBetween(1, 15);
+            return new AzureStorageService(settings, azureClientProvider) {
+                @Override
+                long getUploadBlockSize() {
+                    return blockSize;
+                }
+            };
+        }
+    }
+
     @ClassRule
     public static AzureHttpFixture fixture = new AzureHttpFixture(
         USE_FIXTURE ? AzureHttpFixture.Protocol.HTTP : AzureHttpFixture.Protocol.NONE,
@@ -71,7 +100,7 @@ public class AzureStorageCleanupThirdPartyTests extends AbstractThirdPartyReposi
 
     @Override
     protected Collection<Class<? extends Plugin>> getPlugins() {
-        return pluginList(AzureRepositoryPlugin.class);
+        return pluginList(TestAzureRepositoryPlugin.class);
     }
 
     @Override
@@ -158,19 +187,67 @@ private void ensureSasTokenPermissions() {
 
     public void testMultiBlockUpload() throws Exception {
         final BlobStoreRepository repo = getRepository();
+        assertThat(
+            asInstanceOf(AzureBlobStore.class, repo.blobStore()).getLargeBlobThresholdInBytes(),
+            equalTo(ByteSizeUnit.MB.toBytes(1L))
+        );
+        assertThat(asInstanceOf(AzureBlobStore.class, repo.blobStore()).getUploadBlockSize(), lessThan(ByteSizeUnit.MB.toBytes(1L)));
+
         // The configured threshold for this test suite is 1mb
-        final int blobSize = ByteSizeUnit.MB.toIntBytes(2);
+        final long blobSize = randomLongBetween(ByteSizeUnit.MB.toBytes(2), ByteSizeUnit.MB.toBytes(4));
+        final int bufferSize = 8192;
+
+        final var file = createTempFile();
+        final long expectedChecksum;
+        try (var output = new CheckedOutputStream(new BufferedOutputStream(Files.newOutputStream(file)), new CRC32())) {
+            long remaining = blobSize;
+            while (remaining > 0L) {
+                final var buffer = randomByteArrayOfLength(Math.toIntExact(Math.min(bufferSize, remaining)));
+                output.write(buffer);
+                remaining -= buffer.length;
+            }
+            output.flush();
+            expectedChecksum = output.getChecksum().getValue();
+        }
+
         PlainActionFuture<Void> future = new PlainActionFuture<>();
         repo.threadPool().generic().execute(ActionRunnable.run(future, () -> {
             final BlobContainer blobContainer = repo.blobStore().blobContainer(repo.basePath().add("large_write"));
-            blobContainer.writeBlob(
-                randomPurpose(),
-                UUIDs.base64UUID(),
-                new ByteArrayInputStream(randomByteArrayOfLength(blobSize)),
-                blobSize,
-                false
-            );
-            blobContainer.delete(randomPurpose());
+            try {
+                final var blobName = UUIDs.base64UUID();
+                if (randomBoolean()) {
+                    try (var input = new BufferedInputStream(Files.newInputStream(file))) {
+                        blobContainer.writeBlob(randomPurpose(), blobName, input, blobSize, false);
+                    }
+                } else {
+                    assertThat(blobContainer.supportsConcurrentMultipartUploads(), equalTo(true));
+                    blobContainer.writeBlobAtomic(randomPurpose(), blobName, blobSize, (offset, length) -> {
+                        var channel = Files.newByteChannel(file);
+                        if (offset > 0L) {
+                            if (channel.size() <= offset) {
+                                throw new AssertionError();
+                            }
+                            channel.position(offset);
+                        }
+                        assert channel.position() == offset;
+                        return new BufferedInputStream(limitStream(Channels.newInputStream(channel), length));
+                    }, false);
+                }
+
+                long bytesCount = 0L;
+                try (var input = new CheckedInputStream(blobContainer.readBlob(OperationPurpose.INDICES, blobName), new CRC32())) {
+                    var buffer = new byte[bufferSize];
+                    int bytesRead;
+                    while ((bytesRead = input.read(buffer)) != -1) {
+                        bytesCount += bytesRead;
+                    }
+
+                    assertThat(bytesCount, equalTo(blobSize));
+                    assertThat(input.getChecksum().getValue(), equalTo(expectedChecksum));
+                }
+            } finally {
+                blobContainer.delete(randomPurpose());
+            }
         }));
         future.get();
     }

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

@@ -504,12 +504,10 @@ void writeBlobAtomic(
                     .collect(Collectors.toList())
                     .flatMap(blockIds -> {
                         logger.debug("{}: all {} parts uploaded, now committing", blobName, multiParts.size());
-                        var response = asyncClient.commitBlockList(
+                        return asyncClient.commitBlockList(
                             multiParts.stream().map(MultiPart::blockId).toList(),
                             failIfAlreadyExists == false
-                        );
-                        logger.debug("{}: all {} parts committed", blobName, multiParts.size());
-                        return response;
+                        ).doOnSuccess(unused -> logger.debug("{}: all {} parts committed", blobName, multiParts.size()));
                     })
                     .block();
             }

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

@@ -11,6 +11,7 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Strings;
@@ -110,6 +111,7 @@ public static final class Repository {
     private final RepositoriesMetrics repositoriesMetrics;
 
     public AzureRepository(
+        final ProjectId projectId,
         final RepositoryMetadata metadata,
         final NamedXContentRegistry namedXContentRegistry,
         final AzureStorageService storageService,
@@ -119,6 +121,7 @@ public AzureRepository(
         final RepositoriesMetrics repositoriesMetrics
     ) {
         super(
+            projectId,
             metadata,
             namedXContentRegistry,
             clusterService,

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

@@ -62,10 +62,11 @@ public Map<String, Repository.Factory> getRepositories(
         RecoverySettings recoverySettings,
         RepositoriesMetrics repositoriesMetrics
     ) {
-        return Collections.singletonMap(AzureRepository.TYPE, metadata -> {
+        return Collections.singletonMap(AzureRepository.TYPE, (projectId, metadata) -> {
             AzureStorageService storageService = azureStoreService.get();
             assert storageService != null;
             return new AzureRepository(
+                projectId,
                 metadata,
                 namedXContentRegistry,
                 storageService,

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

@@ -9,6 +9,7 @@
 
 package org.elasticsearch.repositories.azure;
 
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
@@ -23,6 +24,7 @@
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 
 import static org.elasticsearch.repositories.blobstore.BlobStoreRepository.READONLY_SETTING_KEY;
+import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.nullValue;
 import static org.mockito.Mockito.mock;
@@ -35,7 +37,9 @@ private AzureRepository azureRepository(Settings settings) {
             .putList(Environment.PATH_DATA_SETTING.getKey(), tmpPaths())
             .put(settings)
             .build();
+        final ProjectId projectId = randomProjectIdOrDefault();
         final AzureRepository azureRepository = new AzureRepository(
+            projectId,
             new RepositoryMetadata("foo", "azure", internalSettings),
             NamedXContentRegistry.EMPTY,
             mock(AzureStorageService.class),
@@ -44,6 +48,7 @@ private AzureRepository azureRepository(Settings settings) {
             new RecoverySettings(settings, new ClusterSettings(settings, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS)),
             RepositoriesMetrics.NOOP
         );
+        assertThat(azureRepository.getProjectId(), equalTo(projectId));
         assertThat(azureRepository.getBlobStore(), is(nullValue()));
         return azureRepository;
     }

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

@@ -276,7 +276,8 @@ public Map<String, Repository.Factory> getRepositories(
         ) {
             return Collections.singletonMap(
                 GoogleCloudStorageRepository.TYPE,
-                metadata -> new GoogleCloudStorageRepository(
+                (projectId, metadata) -> new GoogleCloudStorageRepository(
+                    projectId,
                     metadata,
                     registry,
                     this.storageService,

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

@@ -57,7 +57,8 @@ public Map<String, Repository.Factory> getRepositories(
     ) {
         return Collections.singletonMap(
             GoogleCloudStorageRepository.TYPE,
-            metadata -> new GoogleCloudStorageRepository(
+            (projectId, metadata) -> new GoogleCloudStorageRepository(
+                projectId,
                 metadata,
                 namedXContentRegistry,
                 this.storageService,

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

@@ -11,6 +11,7 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.BackoffPolicy;
@@ -88,6 +89,7 @@ class GoogleCloudStorageRepository extends MeteredBlobStoreRepository {
     private final GcsRepositoryStatsCollector statsCollector;
 
     GoogleCloudStorageRepository(
+        final ProjectId projectId,
         final RepositoryMetadata metadata,
         final NamedXContentRegistry namedXContentRegistry,
         final GoogleCloudStorageService storageService,
@@ -97,6 +99,7 @@ class GoogleCloudStorageRepository extends MeteredBlobStoreRepository {
         final GcsRepositoryStatsCollector statsCollector
     ) {
         super(
+            projectId,
             metadata,
             namedXContentRegistry,
             clusterService,