Merge remote-tracking branch 'upstream/main' into gcp-metering

Author: mhl-b

modules/repository-s3/src/main/java/org/elasticsearch/repositories/s3/S3BlobStore.java

@@ -55,7 +55,6 @@
 import java.util.concurrent.atomic.LongAdder;
 import java.util.stream.Collectors;
 
-import static org.elasticsearch.core.Strings.format;
 import static org.elasticsearch.rest.RestStatus.REQUESTED_RANGE_NOT_SATISFIED;
 
 class S3BlobStore implements BlobStore {
@@ -398,16 +397,7 @@ private void deletePartition(OperationPurpose purpose, List<String> partition, D
             } catch (MultiObjectDeleteException e) {
                 // We are sending quiet mode requests so we can't use the deleted keys entry on the exception and instead
                 // first remove all keys that were sent in the request and then add back those that ran into an exception.
-                logger.warn(
-                    () -> format(
-                        "Failed to delete some blobs %s",
-                        e.getErrors()
-                            .stream()
-                            .map(err -> "[" + err.getKey() + "][" + err.getCode() + "][" + err.getMessage() + "]")
-                            .toList()
-                    ),
-                    e
-                );
+                logger.warn(buildDeletionErrorMessage(e), e);
                 deletionExceptions.useOrMaybeSuppress(e);
                 return;
             } catch (AmazonClientException e) {
@@ -430,6 +420,26 @@ private void deletePartition(OperationPurpose purpose, List<String> partition, D
         }
     }
 
+    private String buildDeletionErrorMessage(MultiObjectDeleteException e) {
+        final var sb = new StringBuilder("Failed to delete some blobs ");
+        final var errors = e.getErrors();
+        for (int i = 0; i < errors.size() && i < MAX_DELETE_EXCEPTIONS; i++) {
+            final var err = errors.get(i);
+            sb.append("[").append(err.getKey()).append("][").append(err.getCode()).append("][").append(err.getMessage()).append("]");
+            if (i < errors.size() - 1) {
+                sb.append(",");
+            }
+        }
+        if (errors.size() > MAX_DELETE_EXCEPTIONS) {
+            sb.append("... (")
+                .append(errors.size())
+                .append(" in total, ")
+                .append(errors.size() - MAX_DELETE_EXCEPTIONS)
+                .append(" omitted)");
+        }
+        return sb.toString();
+    }
+
     /**
      * If there are remaining retries, pause for the configured interval then return true
      *

modules/repository-s3/src/test/java/org/elasticsearch/repositories/s3/S3BlobContainerRetriesTests.java

@@ -21,6 +21,7 @@
 import com.sun.net.httpserver.HttpHandler;
 
 import org.apache.http.HttpStatus;
+import org.apache.logging.log4j.Level;
 import org.apache.lucene.index.CorruptIndexException;
 import org.apache.lucene.store.AlreadyClosedException;
 import org.elasticsearch.ExceptionsHelper;
@@ -51,10 +52,12 @@
 import org.elasticsearch.repositories.RepositoriesMetrics;
 import org.elasticsearch.repositories.blobstore.AbstractBlobContainerRetriesTestCase;
 import org.elasticsearch.repositories.blobstore.BlobStoreTestUtil;
+import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.telemetry.InstrumentType;
 import org.elasticsearch.telemetry.Measurement;
 import org.elasticsearch.telemetry.RecordingMeterRegistry;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.test.MockLog;
 import org.elasticsearch.watcher.ResourceWatcherService;
 import org.hamcrest.Matcher;
 import org.junit.After;
@@ -65,6 +68,7 @@
 import java.io.FilterInputStream;
 import java.io.IOException;
 import java.io.InputStream;
+import java.io.InputStreamReader;
 import java.net.InetSocketAddress;
 import java.net.SocketTimeoutException;
 import java.net.UnknownHostException;
@@ -83,6 +87,7 @@
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.IntConsumer;
+import java.util.regex.Pattern;
 
 import static org.elasticsearch.repositories.blobstore.BlobStoreTestUtil.randomNonDataPurpose;
 import static org.elasticsearch.repositories.blobstore.BlobStoreTestUtil.randomPurpose;
@@ -1106,6 +1111,60 @@ public void testSuppressedDeletionErrorsAreCapped() {
         assertThat(exception.getCause().getSuppressed().length, lessThan(S3BlobStore.MAX_DELETE_EXCEPTIONS));
     }
 
+    public void testTrimmedLogAndCappedSuppressedErrorOnMultiObjectDeletionException() {
+        final TimeValue readTimeout = TimeValue.timeValueMillis(randomIntBetween(100, 500));
+        int maxBulkDeleteSize = randomIntBetween(10, 30);
+        final BlobContainer blobContainer = createBlobContainer(1, readTimeout, true, null, maxBulkDeleteSize);
+
+        final Pattern pattern = Pattern.compile("<Key>(.+?)</Key>");
+        httpServer.createContext("/", exchange -> {
+            if (exchange.getRequestMethod().equals("POST") && exchange.getRequestURI().toString().startsWith("/bucket/?delete")) {
+                final String requestBody = Streams.copyToString(new InputStreamReader(exchange.getRequestBody(), StandardCharsets.UTF_8));
+                final var matcher = pattern.matcher(requestBody);
+                final StringBuilder deletes = new StringBuilder();
+                deletes.append("<?xml version=\"1.0\" encoding=\"UTF-8\"?>");
+                deletes.append("<DeleteResult>");
+                while (matcher.find()) {
+                    final String key = matcher.group(1);
+                    deletes.append("<Error>");
+                    deletes.append("<Code>").append(randomAlphaOfLength(10)).append("</Code>");
+                    deletes.append("<Key>").append(key).append("</Key>");
+                    deletes.append("<Message>").append(randomAlphaOfLength(40)).append("</Message>");
+                    deletes.append("</Error>");
+                }
+                deletes.append("</DeleteResult>");
+
+                byte[] response = deletes.toString().getBytes(StandardCharsets.UTF_8);
+                exchange.getResponseHeaders().add("Content-Type", "application/xml");
+                exchange.sendResponseHeaders(RestStatus.OK.getStatus(), response.length);
+                exchange.getResponseBody().write(response);
+                exchange.close();
+            } else {
+                fail("expected only deletions");
+            }
+        });
+        var blobs = randomList(maxBulkDeleteSize, maxBulkDeleteSize, ESTestCase::randomIdentifier);
+        try (var mockLog = MockLog.capture(S3BlobStore.class)) {
+            mockLog.addExpectation(
+                new MockLog.SeenEventExpectation(
+                    "deletion log",
+                    S3BlobStore.class.getCanonicalName(),
+                    Level.WARN,
+                    blobs.size() > S3BlobStore.MAX_DELETE_EXCEPTIONS
+                        ? "Failed to delete some blobs [*... (* in total, * omitted)"
+                        : "Failed to delete some blobs [*]"
+                )
+            );
+            var exception = expectThrows(
+                IOException.class,
+                "deletion should not succeed",
+                () -> blobContainer.deleteBlobsIgnoringIfNotExists(randomPurpose(), blobs.iterator())
+            );
+            assertThat(exception.getCause().getSuppressed().length, lessThan(S3BlobStore.MAX_DELETE_EXCEPTIONS));
+            mockLog.awaitAllExpectationsMatched();
+        }
+    }
+
     @Override
     protected Matcher<Integer> getMaxRetriesMatcher(int maxRetries) {
         // some attempts make meaningful progress and do not count towards the max retry limit

muted-tests.yml

@@ -345,6 +345,9 @@ tests:
 - class: org.elasticsearch.compute.data.BlockMultiValuedTests
   method: testToMask {elementType=BOOLEAN nullAllowed=true}
   issue: https://github.com/elastic/elasticsearch/issues/124165
+- class: org.elasticsearch.smoketest.MlWithSecurityIT
+  method: test {yaml=ml/start_data_frame_analytics/Test start classification analysis when the dependent variable is missing}
+  issue: https://github.com/elastic/elasticsearch/issues/124168
 
 # Examples:
 #

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -183,7 +183,8 @@ static TransportVersion def(int id) {
     public static final TransportVersion STORED_SCRIPT_CONTENT_LENGTH = def(9_019_0_00);
     public static final TransportVersion JINA_AI_EMBEDDING_TYPE_SUPPORT_ADDED = def(9_020_0_00);
     public static final TransportVersion RE_REMOVE_MIN_COMPATIBLE_SHARD_NODE = def(9_021_0_00);
-    public static final TransportVersion INCLUDE_INDEX_MODE_IN_GET_DATA_STREAM = def(9_022_0_00);
+    public static final TransportVersion UNASSIGENEDINFO_RESHARD_ADDED = def(9_022_0_00);
+    public static final TransportVersion INCLUDE_INDEX_MODE_IN_GET_DATA_STREAM = def(9_023_0_00);
 
     /*
      * STOP! READ THIS FIRST! No, really,

server/src/main/java/org/elasticsearch/cluster/metadata/IndexMetadata.java

@@ -1947,6 +1947,39 @@ public Builder numberOfShards(int numberOfShards) {
             return this;
         }
 
+        /**
+         * Builder to create IndexMetadata that has an increased shard count (used for re-shard).
+         * The new shard count must be a multiple of the original shardcount.
+         * We do not support shrinking the shard count.
+         * @param shardCount   updated shardCount
+         *
+         * TODO: Check if this.version needs to be incremented
+         */
+        public Builder reshardAddShards(int shardCount) {
+            // Assert routingNumShards is null ?
+            // Assert numberOfShards > 0
+            if (shardCount % numberOfShards() != 0) {
+                throw new IllegalArgumentException(
+                    "New shard count ["
+                        + shardCount
+                        + "] should be a multiple"
+                        + " of current shard count ["
+                        + numberOfShards()
+                        + "] for ["
+                        + index
+                        + "]"
+                );
+            }
+            IndexVersion indexVersionCreated = indexCreatedVersion(settings);
+            settings = Settings.builder().put(settings).put(SETTING_NUMBER_OF_SHARDS, shardCount).build();
+            var newPrimaryTerms = new long[shardCount];
+            Arrays.fill(newPrimaryTerms, this.primaryTerms.length, newPrimaryTerms.length, SequenceNumbers.UNASSIGNED_PRIMARY_TERM);
+            System.arraycopy(primaryTerms, 0, newPrimaryTerms, 0, this.primaryTerms.length);
+            primaryTerms = newPrimaryTerms;
+            routingNumShards = MetadataCreateIndexService.calculateNumRoutingShards(shardCount, indexVersionCreated);
+            return this;
+        }
+
         /**
          * Sets the number of shards that should be used for routing. This should only be used if the number of shards in
          * an index has changed ie if the index is shrunk.

server/src/main/java/org/elasticsearch/cluster/routing/IndexRoutingTable.java

@@ -662,7 +662,7 @@ public Builder addShard(ShardRouting shard) {
             return this;
         }
 
-        void ensureShardArray(int shardCount) {
+        public void ensureShardArray(int shardCount) {
             if (shards == null) {
                 shards = new IndexShardRoutingTable.Builder[shardCount];
             } else if (shards.length < shardCount) {

server/src/main/java/org/elasticsearch/cluster/routing/RoutingTable.java

@@ -436,6 +436,14 @@ public Builder(ShardRoutingRoleStrategy shardRoutingRoleStrategy, RoutingTable r
             this.indicesRouting = ImmutableOpenMap.builder(routingTable.indicesRouting);
         }
 
+        public IndexRoutingTable getIndexRoutingTable(String index) {
+            return indicesRouting.get(index);
+        }
+
+        public ShardRoutingRoleStrategy getShardRoutingRoleStrategy() {
+            return shardRoutingRoleStrategy;
+        }
+
         private static void addShard(
             final Map<String, IndexRoutingTable.Builder> indexRoutingTableBuilders,
             final ShardRouting shardRoutingEntry

server/src/main/java/org/elasticsearch/cluster/routing/UnassignedInfo.java

@@ -175,7 +175,11 @@ public enum Reason {
         /**
          * Replica is unpromotable and the primary failed.
          */
-        UNPROMOTABLE_REPLICA
+        UNPROMOTABLE_REPLICA,
+        /**
+         * New shard added as part of index re-sharding operation
+         */
+        RESHARD_ADDED
     }
 
     /**
@@ -335,9 +339,14 @@ public void writeTo(StreamOutput out) throws IOException {
             out.writeByte((byte) Reason.NODE_LEFT.ordinal());
         } else if (reason.equals(Reason.UNPROMOTABLE_REPLICA) && out.getTransportVersion().before(VERSION_UNPROMOTABLE_REPLICA_ADDED)) {
             out.writeByte((byte) Reason.PRIMARY_FAILED.ordinal());
-        } else {
-            out.writeByte((byte) reason.ordinal());
-        }
+        } else if (reason.equals(Reason.RESHARD_ADDED)
+            && out.getTransportVersion().before(TransportVersions.UNASSIGENEDINFO_RESHARD_ADDED)) {
+                // We should have protection to ensure we do not reshard in mixed clusters
+                assert false;
+                out.writeByte((byte) Reason.FORCED_EMPTY_PRIMARY.ordinal());
+            } else {
+                out.writeByte((byte) reason.ordinal());
+            }
         out.writeLong(unassignedTimeMillis);
         // Do not serialize unassignedTimeNanos as System.nanoTime() cannot be compared across different JVMs
         out.writeBoolean(delayed);

server/src/test/java/org/elasticsearch/cluster/routing/UnassignedInfoTests.java

@@ -88,7 +88,8 @@ public void testReasonOrdinalOrder() {
             UnassignedInfo.Reason.MANUAL_ALLOCATION,
             UnassignedInfo.Reason.INDEX_CLOSED,
             UnassignedInfo.Reason.NODE_RESTARTING,
-            UnassignedInfo.Reason.UNPROMOTABLE_REPLICA };
+            UnassignedInfo.Reason.UNPROMOTABLE_REPLICA,
+            UnassignedInfo.Reason.RESHARD_ADDED };
         for (int i = 0; i < order.length; i++) {
             assertThat(order[i].ordinal(), equalTo(i));
         }