Cross-project search index expression rewriting

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

@@ -80,7 +80,7 @@
 import static org.elasticsearch.cluster.metadata.IndexMetadata.INDEX_UUID_NA_VALUE;
 import static org.elasticsearch.common.xcontent.XContentParserUtils.ensureExpectedToken;
 import static org.elasticsearch.common.xcontent.XContentParserUtils.ensureFieldName;
-import static org.elasticsearch.search.crossproject.IndexExpressionsRewriter.NO_MATCHING_PROJECT_EXCEPTION_VERSION;
+import static org.elasticsearch.search.crossproject.CrossProjectIndexExpressionsRewriter.NO_MATCHING_PROJECT_EXCEPTION_VERSION;
 
 /**
  * A base class for all elasticsearch exceptions.

server/src/main/java/org/elasticsearch/action/IndicesRequest.java

@@ -80,6 +80,13 @@ default ResolvedIndexExpressions getResolvedIndexExpressions() {
         default boolean allowsRemoteIndices() {
             return false;
         }
+
+        /**
+         * Determines whether the request type allows cross-project processing.
+         */
+        default boolean allowsCrossProject() {
+            return false;
+        }
     }
 
     /**

server/src/main/java/org/elasticsearch/action/ResolvedIndexExpression.java

@@ -10,8 +10,12 @@
 package org.elasticsearch.action;
 
 import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.core.Nullable;
 
+import java.io.IOException;
 import java.util.Set;
 
 /**
@@ -40,7 +44,21 @@
  *                         and failure info
  * @param remoteExpressions the remote expressions that replace the original
  */
-public record ResolvedIndexExpression(String original, LocalExpressions localExpressions, Set<String> remoteExpressions) {
+public record ResolvedIndexExpression(String original, LocalExpressions localExpressions, Set<String> remoteExpressions)
+    implements
+        Writeable {
+
+    public ResolvedIndexExpression(StreamInput in) throws IOException {
+        this(in.readString(), new LocalExpressions(in), in.readCollectionAsImmutableSet(StreamInput::readString));
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeString(original);
+        localExpressions.writeTo(out);
+        out.writeStringCollection(remoteExpressions);
+    }
+
     /**
      * Indicates if a local index resolution attempt was successful or failed.
      * Failures can be due to concrete resources not being visible (either missing or not visible due to indices options)
@@ -62,13 +80,28 @@ public record LocalExpressions(
         Set<String> expressions,
         LocalIndexResolutionResult localIndexResolutionResult,
         @Nullable ElasticsearchException exception
-    ) {
+    ) implements Writeable {
         public LocalExpressions {
             assert localIndexResolutionResult != LocalIndexResolutionResult.SUCCESS || exception == null
                 : "If the local resolution result is SUCCESS, exception must be null";
         }
 
         // Singleton for the case where all expressions in a ResolvedIndexExpression instance are remote
         public static final LocalExpressions NONE = new LocalExpressions(Set.of(), LocalIndexResolutionResult.NONE, null);
+
+        public LocalExpressions(StreamInput in) throws IOException {
+            this(
+                in.readCollectionAsImmutableSet(StreamInput::readString),
+                in.readEnum(LocalIndexResolutionResult.class),
+                ElasticsearchException.readException(in)
+            );
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            out.writeStringCollection(expressions);
+            out.writeEnum(localIndexResolutionResult);
+            ElasticsearchException.writeException(exception, out);
+        }
     }
 }

server/src/main/java/org/elasticsearch/action/ResolvedIndexExpressions.java

@@ -9,8 +9,13 @@
 
 package org.elasticsearch.action;
 
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.action.ResolvedIndexExpression.LocalExpressions;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
@@ -20,50 +25,78 @@
 /**
  * A collection of {@link ResolvedIndexExpression}.
  */
-public record ResolvedIndexExpressions(List<ResolvedIndexExpression> expressions) {
+public record ResolvedIndexExpressions(List<ResolvedIndexExpression> expressions) implements Writeable {
+    public static final TransportVersion RESOLVED_INDEX_EXPRESSIONS = TransportVersion.fromName("resolved_index_expressions");
+
+    public ResolvedIndexExpressions(StreamInput in) throws IOException {
+        this(in.readCollectionAsImmutableList(ResolvedIndexExpression::new));
+    }
 
     public List<String> getLocalIndicesList() {
         return expressions.stream().flatMap(e -> e.localExpressions().expressions().stream()).toList();
     }
 
+    public List<String> getRemoteIndicesList() {
+        return expressions.stream().flatMap(e -> e.remoteExpressions().stream()).toList();
+    }
+
     public static Builder builder() {
         return new Builder();
     }
 
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeCollection(expressions);
+    }
+
     public static final class Builder {
         private final List<ResolvedIndexExpression> expressions = new ArrayList<>();
 
         /**
+         * Add a new resolved expression.
          * @param original         the original expression that was resolved -- may be blank for "access all" cases
          * @param localExpressions is a HashSet as an optimization -- the set needs to be mutable, and we want to avoid copying it.
          *                         May be empty.
          */
-        public void addLocalExpressions(
+        public void addExpressions(
             String original,
             HashSet<String> localExpressions,
-            ResolvedIndexExpression.LocalIndexResolutionResult resolutionResult
+            ResolvedIndexExpression.LocalIndexResolutionResult resolutionResult,
+            Set<String> remoteExpressions
         ) {
             Objects.requireNonNull(original);
             Objects.requireNonNull(localExpressions);
             Objects.requireNonNull(resolutionResult);
+            Objects.requireNonNull(remoteExpressions);
             expressions.add(
-                new ResolvedIndexExpression(original, new LocalExpressions(localExpressions, resolutionResult, null), new HashSet<>())
+                new ResolvedIndexExpression(original, new LocalExpressions(localExpressions, resolutionResult, null), remoteExpressions)
             );
         }
 
+        public void addRemoteExpressions(String original, Set<String> remoteExpressions) {
+            Objects.requireNonNull(original);
+            Objects.requireNonNull(remoteExpressions);
+            expressions.add(new ResolvedIndexExpression(original, LocalExpressions.NONE, remoteExpressions));
+        }
+
         /**
          * Exclude the given expressions from the local expressions of all prior added {@link ResolvedIndexExpression}.
          */
         public void excludeFromLocalExpressions(Set<String> expressionsToExclude) {
             Objects.requireNonNull(expressionsToExclude);
             if (expressionsToExclude.isEmpty() == false) {
                 for (ResolvedIndexExpression prior : expressions) {
-                    prior.localExpressions().expressions().removeAll(expressionsToExclude);
+                    final Set<String> localExpressions = prior.localExpressions().expressions();
+                    if (localExpressions.isEmpty()) {
+                        continue;
+                    }
+                    localExpressions.removeAll(expressionsToExclude);
                 }
             }
         }
 
         public ResolvedIndexExpressions build() {
+            // TODO make all sets on `expressions` immutable
             return new ResolvedIndexExpressions(expressions);
         }
     }

server/src/main/java/org/elasticsearch/action/admin/indices/resolve/ResolveIndexAction.java

@@ -9,6 +9,8 @@
 
 package org.elasticsearch.action.admin.indices.resolve;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.elasticsearch.TransportVersion;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRequestValidationException;
@@ -18,6 +20,7 @@
 import org.elasticsearch.action.LegacyActionRequest;
 import org.elasticsearch.action.OriginalIndices;
 import org.elasticsearch.action.RemoteClusterActionType;
+import org.elasticsearch.action.ResolvedIndexExpressions;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.action.support.IndicesOptions;
@@ -43,6 +46,7 @@
 import org.elasticsearch.index.IndexMode;
 import org.elasticsearch.injection.guice.Inject;
 import org.elasticsearch.search.SearchService;
+import org.elasticsearch.search.crossproject.CrossProjectIndexResolutionValidator;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.RemoteClusterAware;
 import org.elasticsearch.transport.RemoteClusterService;
@@ -69,6 +73,8 @@
 import java.util.stream.Stream;
 
 import static org.elasticsearch.action.search.TransportSearchHelper.checkCCSVersionCompatibility;
+import static org.elasticsearch.search.crossproject.CrossProjectIndicesRequestHelper.crossProjectFanoutIndicesOptions;
+import static org.elasticsearch.search.crossproject.CrossProjectIndicesRequestHelper.shouldResolveCrossProject;
 
 public class ResolveIndexAction extends ActionType<ResolveIndexAction.Response> {
 
@@ -90,6 +96,7 @@ public static class Request extends LegacyActionRequest implements IndicesReques
         private String[] names;
         private IndicesOptions indicesOptions = DEFAULT_INDICES_OPTIONS;
         private EnumSet<IndexMode> indexModes = EnumSet.noneOf(IndexMode.class);
+        private ResolvedIndexExpressions resolvedIndexExpressions = null;
 
         public Request(String[] names) {
             this.names = names;
@@ -168,6 +175,21 @@ public boolean allowsRemoteIndices() {
             return true;
         }
 
+        @Override
+        public boolean allowsCrossProject() {
+            return true;
+        }
+
+        @Override
+        public void setResolvedIndexExpressions(ResolvedIndexExpressions expressions) {
+            this.resolvedIndexExpressions = expressions;
+        }
+
+        @Override
+        public ResolvedIndexExpressions getResolvedIndexExpressions() {
+            return resolvedIndexExpressions;
+        }
+
         @Override
         public boolean includeDataStreams() {
             // request must allow data streams because the index name expression resolver for the action handler assumes it
@@ -461,17 +483,34 @@ public static class Response extends ActionResponse implements ToXContentObject
         private final List<ResolvedIndex> indices;
         private final List<ResolvedAlias> aliases;
         private final List<ResolvedDataStream> dataStreams;
+        @Nullable
+        private final ResolvedIndexExpressions resolvedIndexExpressions;
 
         public Response(List<ResolvedIndex> indices, List<ResolvedAlias> aliases, List<ResolvedDataStream> dataStreams) {
+            this(indices, aliases, dataStreams, null);
+        }
+
+        public Response(
+            List<ResolvedIndex> indices,
+            List<ResolvedAlias> aliases,
+            List<ResolvedDataStream> dataStreams,
+            ResolvedIndexExpressions resolvedIndexExpressions
+        ) {
             this.indices = indices;
             this.aliases = aliases;
             this.dataStreams = dataStreams;
+            this.resolvedIndexExpressions = resolvedIndexExpressions;
         }
 
         public Response(StreamInput in) throws IOException {
             this.indices = in.readCollectionAsList(ResolvedIndex::new);
             this.aliases = in.readCollectionAsList(ResolvedAlias::new);
             this.dataStreams = in.readCollectionAsList(ResolvedDataStream::new);
+            if (in.getTransportVersion().supports(ResolvedIndexExpressions.RESOLVED_INDEX_EXPRESSIONS)) {
+                this.resolvedIndexExpressions = in.readOptionalWriteable(ResolvedIndexExpressions::new);
+            } else {
+                this.resolvedIndexExpressions = null;
+            }
         }
 
         public List<ResolvedIndex> getIndices() {
@@ -491,6 +530,9 @@ public void writeTo(StreamOutput out) throws IOException {
             out.writeCollection(indices);
             out.writeCollection(aliases);
             out.writeCollection(dataStreams);
+            if (out.getTransportVersion().supports(ResolvedIndexExpressions.RESOLVED_INDEX_EXPRESSIONS)) {
+                out.writeOptionalWriteable(resolvedIndexExpressions);
+            }
         }
 
         @Override
@@ -515,9 +557,15 @@ public boolean equals(Object o) {
         public int hashCode() {
             return Objects.hash(indices, aliases, dataStreams);
         }
+
+        @Nullable
+        public ResolvedIndexExpressions getResolvedIndexExpressions() {
+            return resolvedIndexExpressions;
+        }
     }
 
     public static class TransportAction extends HandledTransportAction<Request, Response> {
+        private static final Logger logger = LogManager.getLogger(TransportAction.class);
 
         private final ClusterService clusterService;
         private final RemoteClusterService remoteClusterService;
@@ -547,8 +595,10 @@ protected void doExecute(Task task, Request request, final ActionListener<Respon
                 checkCCSVersionCompatibility(request);
             }
             final ProjectState projectState = projectResolver.getProjectState(clusterService.state());
+            final IndicesOptions originalIndicesOptions = request.indicesOptions();
+            final boolean resolveCrossProject = shouldResolveCrossProject(request);
             final Map<String, OriginalIndices> remoteClusterIndices = remoteClusterService.groupIndices(
-                request.indicesOptions(),
+                resolveCrossProject ? crossProjectFanoutIndicesOptions(originalIndicesOptions) : originalIndicesOptions,
                 request.indices()
             );
             final OriginalIndices localIndices = remoteClusterIndices.remove(RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY);
@@ -557,12 +607,34 @@ protected void doExecute(Task task, Request request, final ActionListener<Respon
             List<ResolvedDataStream> dataStreams = new ArrayList<>();
             resolveIndices(localIndices, projectState, indexNameExpressionResolver, indices, aliases, dataStreams, request.indexModes);
 
+            final ResolvedIndexExpressions localResolvedIndexExpressions = request.getResolvedIndexExpressions();
             if (remoteClusterIndices.size() > 0) {
                 final int remoteRequests = remoteClusterIndices.size();
                 final CountDown completionCounter = new CountDown(remoteRequests);
                 final SortedMap<String, Response> remoteResponses = Collections.synchronizedSortedMap(new TreeMap<>());
                 final Runnable terminalHandler = () -> {
                     if (completionCounter.countDown()) {
+                        if (resolveCrossProject) {
+                            // TODO temporary fix: we need to properly handle the case where a remote does not return a result due to
+                            // a failure -- in the current version of resolve indices though, these are just silently ignored
+                            if (remoteRequests != remoteResponses.size()) {
+                                listener.onFailure(
+                                    new IllegalStateException(
+                                        "expected [" + remoteRequests + "] remote responses but got only [" + remoteResponses.size() + "]"
+                                    )
+                                );
+                                return;
+                            }
+                            final Exception ex = CrossProjectIndexResolutionValidator.validate(
+                                originalIndicesOptions,
+                                localResolvedIndexExpressions,
+                                getResolvedExpressionsByRemote(remoteResponses)
+                            );
+                            if (ex != null) {
+                                listener.onFailure(ex);
+                                return;
+                            }
+                        }
                         mergeResults(remoteResponses, indices, aliases, dataStreams, request.indexModes);
                         listener.onResponse(new Response(indices, aliases, dataStreams));
                     }
@@ -581,13 +653,38 @@ protected void doExecute(Task task, Request request, final ActionListener<Respon
                     remoteClusterClient.execute(ResolveIndexAction.REMOTE_TYPE, remoteRequest, ActionListener.wrap(response -> {
                         remoteResponses.put(clusterAlias, response);
                         terminalHandler.run();
-                    }, failure -> terminalHandler.run()));
+                    }, failure -> {
+                        logger.info("failed to resolve indices on remote cluster [" + clusterAlias + "]", failure);
+                        terminalHandler.run();
+                    }));
                 }
             } else {
-                listener.onResponse(new Response(indices, aliases, dataStreams));
+                if (resolveCrossProject) {
+                    // we still need to call response validation for local results, since qualified expressions like `_origin:index` or
+                    // `<alias-pattern-matching-origin-only>:index` get deferred validation, also
+                    final Exception ex = CrossProjectIndexResolutionValidator.validate(
+                        originalIndicesOptions,
+                        localResolvedIndexExpressions,
+                        Map.of()
+                    );
+                    if (ex != null) {
+                        listener.onFailure(ex);
+                        return;
+                    }
+                }
+                listener.onResponse(new Response(indices, aliases, dataStreams, localResolvedIndexExpressions));
             }
         }
 
+        private Map<String, ResolvedIndexExpressions> getResolvedExpressionsByRemote(Map<String, Response> remoteResponses) {
+            return remoteResponses.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, e -> {
+                final ResolvedIndexExpressions resolvedIndexExpressions = e.getValue().getResolvedIndexExpressions();
+                assert resolvedIndexExpressions != null
+                    : "remote response from cluster [" + e.getKey() + "] is missing resolved index expressions";
+                return resolvedIndexExpressions;
+            }));
+        }
+
         /**
          * Resolves the specified names and/or wildcard expressions to index abstractions. Returns results in the supplied lists.
          *