Merge branch 'main' of https://github.com/elastic/elasticsearch into esql-reranker-boostrap

Author: afoucret

docs/changelog/124574.yaml

@@ -0,0 +1,5 @@
+pr: 124574
+summary: Allow passing several reserved state chunks in single process call
+area: Infra/Settings
+type: enhancement
+issues: []

muted-tests.yml

@@ -390,9 +390,6 @@ tests:
 - class: org.elasticsearch.backwards.MixedClusterClientYamlTestSuiteIT
   method: test {p0=search/610_function_score/Random}
   issue: https://github.com/elastic/elasticsearch/issues/125010
-- class: org.elasticsearch.packaging.test.DockerTests
-  method: test010Install
-  issue: https://github.com/elastic/elasticsearch/issues/119441
 - class: org.elasticsearch.xpack.ilm.DataStreamAndIndexLifecycleMixingTests
   method: testGetDataStreamResponse
   issue: https://github.com/elastic/elasticsearch/issues/125083

plugins/examples/security-authorization-engine/src/main/java/org/elasticsearch/example/CustomAuthorizationEngine.java

@@ -10,6 +10,7 @@
 package org.elasticsearch.example;
 
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.support.IndexComponentSelector;
 import org.elasticsearch.action.support.SubscribableListener;
 import org.elasticsearch.cluster.metadata.IndexAbstraction;
 import org.elasticsearch.cluster.metadata.ProjectMetadata;
@@ -35,7 +36,6 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.function.Supplier;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -119,19 +119,19 @@ public void loadAuthorizedIndices(
     ) {
         if (isSuperuser(requestInfo.getAuthentication().getEffectiveSubject().getUser())) {
             listener.onResponse(new AuthorizedIndices() {
-                public Supplier<Set<String>> all() {
+                public Set<String> all(IndexComponentSelector selector) {
                     return () -> indicesLookup.keySet();
                 }
-                public boolean check(String name) {
+                public boolean check(String name, IndexComponentSelector selector) {
                     return indicesLookup.containsKey(name);
                 }
             });
         } else {
             listener.onResponse(new AuthorizedIndices() {
-                public Supplier<Set<String>> all() {
+                public Set<String> all(IndexComponentSelector selector) {
                     return () -> Set.of();
                 }
-                public boolean check(String name) {
+                public boolean check(String name, IndexComponentSelector selector) {
                     return false;
                 }
             });

qa/packaging/src/test/java/org/elasticsearch/packaging/util/docker/Docker.java

@@ -187,7 +187,7 @@ public static void waitForElasticsearchToStart() {
                 Thread.sleep(STARTUP_SLEEP_INTERVAL_MILLISECONDS);
 
                 // Set COLUMNS so that `ps` doesn't truncate its output
-                psOutput = dockerShell.run("bash -c 'COLUMNS=3000 ps ax'").stdout();
+                psOutput = dockerShell.run("bash -c 'COLUMNS=4000 ps ax'").stdout();
 
                 if (psOutput.contains("org.elasticsearch.bootstrap.Elasticsearch")) {
                     isElasticsearchRunning = true;

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

@@ -149,6 +149,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion INFERENCE_CONTEXT_8_X = def(8_841_0_08);
     public static final TransportVersion ML_INFERENCE_DEEPSEEK_8_19 = def(8_841_0_09);
     public static final TransportVersion ESQL_SERIALIZE_BLOCK_TYPE_CODE_8_19 = def(8_841_0_10);
+    public static final TransportVersion ESQL_FAILURE_FROM_REMOTE_8_19 = def(8_841_0_11);
     public static final TransportVersion INITIAL_ELASTICSEARCH_9_0 = def(9_000_0_00);
     public static final TransportVersion REMOVE_SNAPSHOT_FAILURES_90 = def(9_000_0_01);
     public static final TransportVersion TRANSPORT_STATS_HANDLING_TIME_REQUIRED_90 = def(9_000_0_02);

server/src/main/java/org/elasticsearch/action/support/IndexComponentSelector.java

@@ -72,6 +72,24 @@ public static IndexComponentSelector getByKey(String key) {
         return KEY_REGISTRY.get(key);
     }
 
+    /**
+     * Like {@link #getByKey(String)} but throws an exception if the key is not recognised.
+     * @return the selector if recognized. `null` input will return `DATA`.
+     * @throws IllegalArgumentException if the key was not recognised.
+     */
+    public static IndexComponentSelector getByKeyOrThrow(@Nullable String key) {
+        if (key == null) {
+            return DATA;
+        }
+        IndexComponentSelector selector = getByKey(key);
+        if (selector == null) {
+            throw new IllegalArgumentException(
+                "Unknown key of index component selector [" + key + "], available options are: " + KEY_REGISTRY.keySet()
+            );
+        }
+        return selector;
+    }
+
     public static IndexComponentSelector read(StreamInput in) throws IOException {
         byte id = in.readByte();
         if (in.getTransportVersion().onOrAfter(TransportVersions.REMOVE_ALL_APPLICABLE_SELECTOR)

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

@@ -100,6 +100,13 @@ default boolean isDataStreamRelated() {
         return false;
     }
 
+    /**
+     * @return whether this index abstraction is a failure index of a data stream
+     */
+    default boolean isFailureIndexOfDataStream() {
+        return false;
+    }
+
     /**
      * An index abstraction type.
      */
@@ -183,6 +190,11 @@ public DataStream getParentDataStream() {
             return dataStream;
         }
 
+        @Override
+        public boolean isFailureIndexOfDataStream() {
+            return getParentDataStream() != null && getParentDataStream().isFailureStoreIndex(getName());
+        }
+
         @Override
         public boolean isHidden() {
             return isHidden;

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

@@ -22,8 +22,8 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
-import java.util.function.Predicate;
-import java.util.function.Supplier;
+import java.util.function.BiPredicate;
+import java.util.function.Function;
 
 public class IndexAbstractionResolver {
 
@@ -37,8 +37,8 @@ public List<String> resolveIndexAbstractions(
         Iterable<String> indices,
         IndicesOptions indicesOptions,
         ProjectMetadata projectMetadata,
-        Supplier<Set<String>> allAuthorizedAndAvailable,
-        Predicate<String> isAuthorized,
+        Function<IndexComponentSelector, Set<String>> allAuthorizedAndAvailableBySelector,
+        BiPredicate<String, IndexComponentSelector> isAuthorized,
         boolean includeDataStreams
     ) {
         List<String> finalIndices = new ArrayList<>();
@@ -64,14 +64,15 @@ public List<String> resolveIndexAbstractions(
                 );
             }
             indexAbstraction = expressionAndSelector.v1();
+            IndexComponentSelector selector = IndexComponentSelector.getByKeyOrThrow(selectorString);
 
             // we always need to check for date math expressions
             indexAbstraction = IndexNameExpressionResolver.resolveDateMathExpression(indexAbstraction);
 
             if (indicesOptions.expandWildcardExpressions() && Regex.isSimpleMatchPattern(indexAbstraction)) {
                 wildcardSeen = true;
                 Set<String> resolvedIndices = new HashSet<>();
-                for (String authorizedIndex : allAuthorizedAndAvailable.get()) {
+                for (String authorizedIndex : allAuthorizedAndAvailableBySelector.apply(selector)) {
                     if (Regex.simpleMatch(indexAbstraction, authorizedIndex)
                         && isIndexVisible(
                             indexAbstraction,
@@ -102,7 +103,7 @@ && isIndexVisible(
                 resolveSelectorsAndCollect(indexAbstraction, selectorString, indicesOptions, resolvedIndices, projectMetadata);
                 if (minus) {
                     finalIndices.removeAll(resolvedIndices);
-                } else if (indicesOptions.ignoreUnavailable() == false || isAuthorized.test(indexAbstraction)) {
+                } else if (indicesOptions.ignoreUnavailable() == false || isAuthorized.test(indexAbstraction, selector)) {
                     // Unauthorized names are considered unavailable, so if `ignoreUnavailable` is `true` they should be silently
                     // discarded from the `finalIndices` list. Other "ways of unavailable" must be handled by the action
                     // handler, see: https://github.com/elastic/elasticsearch/issues/90215

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

@@ -28,6 +28,7 @@
 import org.elasticsearch.common.util.CollectionUtils;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.util.set.Sets;
+import org.elasticsearch.core.Assertions;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.Predicates;
 import org.elasticsearch.core.Tuple;
@@ -1001,6 +1002,14 @@ public static boolean hasSelectorSuffix(String expression) {
         return expression.contains(SelectorResolver.SELECTOR_SEPARATOR);
     }
 
+    public static boolean hasSelector(@Nullable String expression, IndexComponentSelector selector) {
+        Objects.requireNonNull(selector, "null selectors not supported");
+        if (expression == null) {
+            return false;
+        }
+        return expression.endsWith(SelectorResolver.SELECTOR_SEPARATOR + selector.getKey());
+    }
+
     /**
      * @return If the specified string is a selector expression then this method returns the base expression and its selector part.
      */
@@ -1022,6 +1031,14 @@ public static String combineSelectorExpression(String baseExpression, @Nullable
             : (baseExpression + SelectorResolver.SELECTOR_SEPARATOR + selectorExpression);
     }
 
+    public static void assertExpressionHasNullOrDataSelector(String expression) {
+        if (Assertions.ENABLED) {
+            var tuple = splitSelectorExpression(expression);
+            assert tuple.v2() == null || IndexComponentSelector.DATA.getKey().equals(tuple.v2())
+                : "Expected expression [" + expression + "] to have a data selector but found [" + tuple.v2() + "]";
+        }
+    }
+
     /**
      * Resolve an array of expressions to the set of indices and aliases that these expressions match.
      */

server/src/main/java/org/elasticsearch/reservedstate/service/ReservedClusterStateService.java

@@ -22,6 +22,7 @@
 import org.elasticsearch.cluster.routing.RerouteService;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Priority;
+import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.core.FixForMultiProject;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.env.BuildVersion;
@@ -187,7 +188,7 @@ public void process(
         process(namespace, stateChunk, versionCheck, errorListener);
     }
 
-    ReservedStateChunk parse(ProjectId projectId, String namespace, XContentParser parser) {
+    public ReservedStateChunk parse(ProjectId projectId, String namespace, XContentParser parser) {
         try {
             return stateChunkParser.apply(parser, null);
         } catch (Exception e) {
@@ -377,6 +378,7 @@ public void onFailure(Exception e) {
      * @param projectId the project state to modify
      * @param namespace the namespace under which we'll store the reserved keys in the cluster state metadata
      * @param reservedStateChunk a {@link ReservedStateChunk} composite state object to process
+     * @param versionCheck  Enum representing whether a reserved state should be processed based on the current and new versions
      * @param errorListener a consumer called with {@link IllegalStateException} if the content has errors and the
      *        cluster state cannot be correctly applied, null if successful or the state failed to apply because of incompatible version.
      */
@@ -387,17 +389,41 @@ public void process(
         ReservedStateVersionCheck versionCheck,
         Consumer<Exception> errorListener
     ) {
-        Map<String, Object> reservedState = reservedStateChunk.state();
-        ReservedStateVersion reservedStateVersion = reservedStateChunk.metadata();
+        process(projectId, namespace, List.of(reservedStateChunk), versionCheck, errorListener);
+    }
 
+    /**
+     * Saves and reserves a chunk of the cluster state under a given 'namespace' from {@link XContentParser} by combining several chunks
+     * into one
+     *
+     * @param projectId the project state to modify
+     * @param namespace the namespace under which we'll store the reserved keys in the cluster state metadata
+     * @param reservedStateChunks a list of {@link ReservedStateChunk} composite state objects to process
+     * @param versionCheck  Enum representing whether a reserved state should be processed based on the current and new versions
+     * @param errorListener a consumer called with {@link IllegalStateException} if the content has errors and the
+     *        cluster state cannot be correctly applied, null if successful or the state failed to apply because of incompatible version.
+     */
+    public void process(
+        ProjectId projectId,
+        String namespace,
+        List<ReservedStateChunk> reservedStateChunks,
+        ReservedStateVersionCheck versionCheck,
+        Consumer<Exception> errorListener
+    ) {
+        ReservedStateChunk reservedStateChunk;
+        ReservedStateVersion reservedStateVersion;
         LinkedHashSet<String> orderedHandlers;
+
         try {
+            reservedStateChunk = mergeReservedStateChunks(reservedStateChunks);
+            Map<String, Object> reservedState = reservedStateChunk.state();
+            reservedStateVersion = reservedStateChunk.metadata();
             orderedHandlers = orderedProjectStateHandlers(reservedState.keySet());
         } catch (Exception e) {
             ErrorState errorState = new ErrorState(
                 projectId,
                 namespace,
-                reservedStateVersion.version(),
+                reservedStateChunks.getFirst().metadata().version(),
                 versionCheck,
                 e,
                 ReservedStateErrorMetadata.ErrorKind.PARSING
@@ -476,6 +502,36 @@ public void onFailure(Exception e) {
         );
     }
 
+    private static ReservedStateChunk mergeReservedStateChunks(List<ReservedStateChunk> chunks) {
+        if (chunks.isEmpty()) {
+            throw new IllegalArgumentException("No chunks provided");
+        }
+
+        if (chunks.size() == 1) {
+            return chunks.getFirst();
+        }
+
+        ReservedStateVersion reservedStateVersion = chunks.getFirst().metadata();
+        Map<String, Object> mergedChunks = new HashMap<>(chunks.size());
+        for (var chunk : chunks) {
+            Set<String> duplicateKeys = Sets.intersection(chunk.state().keySet(), mergedChunks.keySet());
+            if (chunk.metadata().equals(reservedStateVersion) == false) {
+                throw new IllegalStateException(
+                    "Failed to merge reserved state chunks because of version mismatch: ["
+                        + reservedStateVersion
+                        + "] != ["
+                        + chunk.metadata()
+                        + "]"
+                );
+            } else if (duplicateKeys.isEmpty() == false) {
+                throw new IllegalStateException("Failed to merge reserved state chunks because of duplicate keys: " + duplicateKeys);
+            }
+            mergedChunks.putAll(chunk.state());
+        }
+
+        return new ReservedStateChunk(mergedChunks, reservedStateVersion);
+    }
+
     // package private for testing
     Exception checkAndReportError(
         Optional<ProjectId> projectId,
@@ -518,6 +574,11 @@ void updateErrorState(ErrorState errorState) {
             return;
         }
 
+        if (errorState.projectId().isPresent() && clusterService.state().metadata().hasProject(errorState.projectId().get()) == false) {
+            // Can't update error state for a project that doesn't exist yet
+            return;
+        }
+
         submitErrorUpdateTask(errorState);
     }