POC: Cross-Project Search

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

@@ -0,0 +1,44 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch;
+
+import org.elasticsearch.action.IndicesRequest;
+import org.elasticsearch.core.Nullable;
+
+import java.util.List;
+
+public interface CrossProjectAwareRequest extends IndicesRequest {
+    /**
+     * Can be used to determine if this should be processed in cross-project mode vs. stateful CCS.
+     */
+    boolean crossProjectModeEnabled();
+
+    /**
+     * Only called if cross-project rewriting (flat-world, linked project filtering) was applied
+     */
+    void qualified(List<QualifiedExpression> qualifiedExpressions);
+
+    @Nullable
+    String queryRouting();
+
+    /**
+     * Used to track a mapping from original expression (potentially flat-world) to canonicalized CCS expressions.
+     * e.g. for an original index expression `logs-*`, this would be:
+     * original=logs-*
+     * qualified=[(logs-*, _local), (my-remote:logs-*, my-remote)]
+     */
+    record QualifiedExpression(String original, List<ExpressionWithProject> qualified) {
+        public boolean hasFlatOriginalExpression() {
+            return true;
+        }
+    }
+
+    record ExpressionWithProject(String expression, String project) {}
+}

server/src/main/java/org/elasticsearch/action/fieldcaps/FieldCapabilitiesRequest.java

@@ -9,6 +9,7 @@
 
 package org.elasticsearch.action.fieldcaps;
 
+import org.elasticsearch.CrossProjectAwareRequest;
 import org.elasticsearch.TransportVersions;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.IndicesRequest;
@@ -36,11 +37,16 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 
-public final class FieldCapabilitiesRequest extends LegacyActionRequest implements IndicesRequest.Replaceable, ToXContentObject {
+public final class FieldCapabilitiesRequest extends LegacyActionRequest
+    implements
+        CrossProjectAwareRequest,
+        IndicesRequest.Replaceable,
+        ToXContentObject {
     public static final String NAME = "field_caps_request";
     public static final IndicesOptions DEFAULT_INDICES_OPTIONS = IndicesOptions.strictExpandOpenAndForbidClosed();
 
@@ -58,6 +64,7 @@ public final class FieldCapabilitiesRequest extends LegacyActionRequest implemen
     private QueryBuilder indexFilter;
     private Map<String, Object> runtimeFields = Collections.emptyMap();
     private Long nowInMillis;
+    private List<QualifiedExpression> qualifiedExpressions;
 
     public FieldCapabilitiesRequest(StreamInput in) throws IOException {
         super(in);
@@ -373,4 +380,25 @@ public String getDescription() {
             }
         };
     }
+
+    @Override
+    public boolean crossProjectModeEnabled() {
+        return qualifiedExpressions != null;
+    }
+
+    @Override
+    public void qualified(List<QualifiedExpression> qualifiedExpressions) {
+        this.qualifiedExpressions = qualifiedExpressions;
+        indices(
+            qualifiedExpressions.stream()
+                .flatMap(indexExpression -> indexExpression.qualified().stream().map(ExpressionWithProject::expression))
+                .toArray(String[]::new)
+        );
+    }
+
+    @Override
+    public String queryRouting() {
+        // TODO how would this look in ES|QL?
+        return null;
+    }
 }

server/src/main/java/org/elasticsearch/action/search/SearchRequest.java

@@ -9,6 +9,7 @@
 
 package org.elasticsearch.action.search;
 
+import org.elasticsearch.CrossProjectAwareRequest;
 import org.elasticsearch.TransportVersions;
 import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionRequestValidationException;
@@ -53,7 +54,11 @@
  * @see Client#search(SearchRequest)
  * @see SearchResponse
  */
-public class SearchRequest extends LegacyActionRequest implements IndicesRequest.Replaceable, Rewriteable<SearchRequest> {
+public class SearchRequest extends LegacyActionRequest
+    implements
+        CrossProjectAwareRequest,
+        IndicesRequest.Replaceable,
+        Rewriteable<SearchRequest> {
 
     public static final ToXContent.Params FORMAT_PARAMS = new ToXContent.MapParams(Collections.singletonMap("pretty", "false"));
 
@@ -70,6 +75,12 @@ public class SearchRequest extends LegacyActionRequest implements IndicesRequest
 
     private String[] indices = Strings.EMPTY_ARRAY;
 
+    @Nullable
+    private String queryRouting = null;
+
+    @Nullable
+    private List<QualifiedExpression> qualifiedExpressions;
+
     @Nullable
     private String routing;
     @Nullable
@@ -400,6 +411,11 @@ public SearchRequest indices(String... indices) {
         return this;
     }
 
+    public SearchRequest queryRouting(String queryRouting) {
+        this.queryRouting = queryRouting;
+        return this;
+    }
+
     private static void validateIndices(String... indices) {
         Objects.requireNonNull(indices, "indices must not be null");
         for (String index : indices) {
@@ -853,4 +869,24 @@ public String toString() {
             + source
             + '}';
     }
+
+    @Override
+    public boolean crossProjectModeEnabled() {
+        return qualifiedExpressions != null;
+    }
+
+    @Override
+    public void qualified(List<QualifiedExpression> qualifiedExpressions) {
+        this.qualifiedExpressions = qualifiedExpressions;
+        indices(
+            qualifiedExpressions.stream()
+                .flatMap(indexExpression -> indexExpression.qualified().stream().map(ExpressionWithProject::expression))
+                .toArray(String[]::new)
+        );
+    }
+
+    @Override
+    public String queryRouting() {
+        return queryRouting;
+    }
 }

server/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java

@@ -367,6 +367,7 @@ public void apply(Settings value, Settings current, Settings previous) {
         TransportSearchAction.SHARD_COUNT_LIMIT_SETTING,
         TransportSearchAction.DEFAULT_PRE_FILTER_SHARD_SIZE,
         RemoteClusterService.REMOTE_CLUSTER_SKIP_UNAVAILABLE,
+        RemoteClusterService.REMOTE_CLUSTER_TAGS,
         SniffConnectionStrategy.REMOTE_CONNECTIONS_PER_CLUSTER,
         RemoteClusterService.REMOTE_INITIAL_CONNECTION_TIMEOUT_SETTING,
         RemoteClusterService.REMOTE_NODE_ATTRIBUTE,

server/src/main/java/org/elasticsearch/rest/action/search/RestSearchAction.java

@@ -9,6 +9,8 @@
 
 package org.elasticsearch.rest.action.search;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.search.SearchRequest;
@@ -63,6 +65,7 @@ public class RestSearchAction extends BaseRestHandler {
     public static final String TYPED_KEYS_PARAM = "typed_keys";
     public static final String INCLUDE_NAMED_QUERIES_SCORE_PARAM = "include_named_queries_score";
     public static final Set<String> RESPONSE_PARAMS = Set.of(TYPED_KEYS_PARAM, TOTAL_HITS_AS_INT_PARAM, INCLUDE_NAMED_QUERIES_SCORE_PARAM);
+    private static final Logger log = LogManager.getLogger(RestSearchAction.class);
 
     private final SearchUsageHolder searchUsageHolder;
     private final Predicate<NodeFeature> clusterSupportsFeature;
@@ -98,6 +101,7 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
             client.threadPool().getThreadContext().setErrorTraceTransportHeader(request);
         }
         SearchRequest searchRequest = new SearchRequest();
+
         // access the BwC param, but just drop it
         // this might be set by old clients
         request.param("min_compatible_shard_node");
@@ -167,6 +171,14 @@ public static void parseSearchRequest(
             searchRequest.source(new SearchSourceBuilder());
         }
         searchRequest.indices(Strings.splitStringByCommaToArray(request.param("index")));
+
+        String queryRouting = request.param("query_routing", null);
+        if (queryRouting != null) {
+            searchRequest.queryRouting(queryRouting);
+        } else {
+            log.info("No query routing defined");
+        }
+
         if (requestContentParser != null) {
             if (searchUsageHolder == null) {
                 searchRequest.source().parseXContent(requestContentParser, true, clusterSupportsFeature);

server/src/main/java/org/elasticsearch/transport/RemoteClusterService.java

@@ -41,6 +41,7 @@
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
@@ -97,6 +98,33 @@ public final class RemoteClusterService extends RemoteClusterAware
         (ns, key) -> boolSetting(key, true, new RemoteConnectionEnabled<>(ns, key), Setting.Property.Dynamic, Setting.Property.NodeScope)
     );
 
+    public record Metadata(String key, String value) {
+        public static Metadata fromString(String tag) {
+            if (tag == null || tag.isEmpty()) {
+                throw new IllegalArgumentException("Remote tag must not be null or empty");
+            }
+            // - as a separator to simplify search path param parsing; won't be like this in the real implementation
+            int idx = tag.indexOf('-');
+            if (idx < 0) {
+                return new Metadata(tag, "");
+            } else {
+                return new Metadata(tag.substring(0, idx), tag.substring(idx + 1));
+            }
+        }
+    }
+
+    public static final Setting.AffixSetting<List<Metadata>> REMOTE_CLUSTER_TAGS = Setting.affixKeySetting(
+        "cluster.remote.",
+        "tags",
+        (ns, key) -> Setting.listSetting(
+            key,
+            Collections.emptyList(),
+            Metadata::fromString,
+            Setting.Property.Dynamic,
+            Setting.Property.NodeScope
+        )
+    );
+
     public static final Setting.AffixSetting<TimeValue> REMOTE_CLUSTER_PING_SCHEDULE = Setting.affixKeySetting(
         "cluster.remote.",
         "transport.ping_schedule",

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/CrossProjectRemoteServerTransportInterceptor.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.core.security;
+
+import org.elasticsearch.transport.Transport;
+import org.elasticsearch.transport.TransportInterceptor;
+import org.elasticsearch.transport.TransportRequest;
+import org.elasticsearch.transport.TransportRequestOptions;
+import org.elasticsearch.transport.TransportResponse;
+import org.elasticsearch.transport.TransportResponseHandler;
+
+public interface CrossProjectRemoteServerTransportInterceptor {
+    // TODO probably don't want this
+    boolean enabled();
+
+    // TODO this should be a wrapper around TransportInterceptor.AsyncSender instead
+    <T extends TransportResponse> void sendRequest(
+        TransportInterceptor.AsyncSender sender,
+        Transport.Connection connection,
+        String action,
+        TransportRequest request,
+        TransportRequestOptions options,
+        TransportResponseHandler<T> handler
+    );
+
+    CustomServerTransportFilter getFilter();
+
+    class Default implements CrossProjectRemoteServerTransportInterceptor {
+        @Override
+        public boolean enabled() {
+            return false;
+        }
+
+        @Override
+        public <T extends TransportResponse> void sendRequest(
+            TransportInterceptor.AsyncSender sender,
+            Transport.Connection connection,
+            String action,
+            TransportRequest request,
+            TransportRequestOptions options,
+            TransportResponseHandler<T> handler
+        ) {
+            sender.sendRequest(connection, action, request, options, handler);
+        }
+
+        @Override
+        public CustomServerTransportFilter getFilter() {
+            return new CustomServerTransportFilter.Default();
+        }
+    }
+}

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/CustomServerTransportFilter.java

@@ -0,0 +1,22 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.core.security;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.transport.TransportRequest;
+
+public interface CustomServerTransportFilter {
+    void filter(String securityAction, TransportRequest request, ActionListener<Void> authenticationListener);
+
+    class Default implements CustomServerTransportFilter {
+        @Override
+        public void filter(String securityAction, TransportRequest request, ActionListener<Void> listener) {
+            listener.onResponse(null);
+        }
+    }
+}

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/SecurityExtension.java

@@ -21,6 +21,7 @@
 import org.elasticsearch.xpack.core.security.authc.service.ServiceAccountTokenStore;
 import org.elasticsearch.xpack.core.security.authc.support.UserRoleMapper;
 import org.elasticsearch.xpack.core.security.authz.AuthorizationEngine;
+import org.elasticsearch.xpack.core.security.authz.CrossProjectTargetResolver;
 import org.elasticsearch.xpack.core.security.authz.RoleDescriptor;
 import org.elasticsearch.xpack.core.security.authz.store.RoleRetrievalResult;
 
@@ -133,6 +134,14 @@ default CustomApiKeyAuthenticator getCustomApiKeyAuthenticator(SecurityComponent
         return null;
     }
 
+    default CrossProjectTargetResolver getCrossProjectTargetResolver(SecurityComponents components) {
+        return null;
+    }
+
+    default CrossProjectRemoteServerTransportInterceptor getCustomRemoteServerTransportInterceptor(SecurityComponents components) {
+        return null;
+    }
+
     /**
      * Returns a authorization engine for authorizing requests, or null to use the default authorization mechanism.
      *