POC: Cross-Project Search

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

@@ -0,0 +1,37 @@
+/*
+ * 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.action;
+
+import org.elasticsearch.core.Nullable;
+
+import java.util.List;
+
+/**
+ * Returns authorized projects (linked and origin), if running in a cross-project environment.
+ * In non-cross-project environments, returns a special value indicating that cross-project does not apply at all.
+ */
+public interface AuthorizedProjectsSupplier {
+    AuthorizedProjects get();
+
+    class Default implements AuthorizedProjectsSupplier {
+        @Override
+        public AuthorizedProjects get() {
+            return AuthorizedProjects.NOT_CROSS_PROJECT;
+        }
+    }
+
+    record AuthorizedProjects(@Nullable String origin, List<String> projects) {
+        public static AuthorizedProjects NOT_CROSS_PROJECT = new AuthorizedProjects(null, List.of());
+
+        public boolean isOriginOnly() {
+            return origin != null && projects.isEmpty();
+        }
+    }
+}

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

@@ -0,0 +1,79 @@
+/*
+ * 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.action;
+
+import org.elasticsearch.transport.RemoteClusterAware;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+public final class CrossProjectReplacedIndexExpressions extends ReplacedIndexExpressions {
+
+    public CrossProjectReplacedIndexExpressions(Map<String, ReplacedIndexExpression> replacedExpressionMap) {
+        super(replacedExpressionMap);
+    }
+
+    public static boolean isQualifiedIndexExpression(String indexExpression) {
+        return RemoteClusterAware.isRemoteIndexName(indexExpression);
+    }
+
+    public List<String> getLocalExpressions() {
+        return replacedExpressionMap.values()
+            .stream()
+            .filter(e -> hasCanonicalExpressionForOrigin(e.replacedBy()))
+            .map(ReplacedIndexExpression::original)
+            .toList();
+    }
+
+    public static boolean hasCanonicalExpressionForOrigin(List<String> replacedBy) {
+        return replacedBy.stream().anyMatch(e -> false == isQualifiedIndexExpression(e));
+    }
+
+    public void replaceLocalExpressions(ReplacedIndexExpressions localResolved) {
+        if (localResolved == null || localResolved.asMap() == null || localResolved.asMap().isEmpty()) {
+            return;
+        }
+
+        for (Map.Entry<String, ReplacedIndexExpression> e : localResolved.asMap().entrySet()) {
+            final String original = e.getKey();
+            final ReplacedIndexExpression local = e.getValue();
+            if (local == null) {
+                continue;
+            }
+
+            final ReplacedIndexExpression current = replacedExpressionMap.get(original);
+            if (current == null) {
+                continue;
+            }
+
+            final List<String> qualified = current.replacedBy()
+                .stream()
+                .filter(org.elasticsearch.action.CrossProjectReplacedIndexExpressions::isQualifiedIndexExpression)
+                .toList();
+
+            final List<String> resolvedLocal = local.replacedBy();
+
+            final List<String> combined = new ArrayList<>(resolvedLocal.size() + qualified.size());
+            combined.addAll(resolvedLocal);
+            combined.addAll(qualified);
+
+            replacedExpressionMap.put(
+                original,
+                new ReplacedIndexExpression(original, combined, local.authorized(), local.existsAndVisible(), null)
+            );
+        }
+    }
+
+    @Override
+    public String toString() {
+        return "CrossProjectReplacedIndexExpressions[" + "replacedExpressionMap=" + replacedExpressionMap + ']';
+    }
+}

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

@@ -9,10 +9,22 @@
 
 package org.elasticsearch.action;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.ElasticsearchSecurityException;
 import org.elasticsearch.action.support.IndicesOptions;
+import org.elasticsearch.core.Nullable;
+import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.rest.RestStatus;
 
+import java.util.ArrayList;
 import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+
+import static org.elasticsearch.action.CrossProjectReplacedIndexExpressions.hasCanonicalExpressionForOrigin;
 
 /**
  * Needs to be implemented by all {@link org.elasticsearch.action.ActionRequest} subclasses that relate to
@@ -48,9 +60,15 @@ interface Replaceable extends IndicesRequest {
          */
         IndicesRequest indices(String... indices);
 
+        default void setReplacedIndexExpressions(ReplacedIndexExpressions replacedIndexExpressions) {}
+
+        @Nullable
+        default ReplacedIndexExpressions getReplacedIndexExpressions() {
+            return null;
+        }
+
         /**
          * Determines whether the request can contain indices on a remote cluster.
-         *
          * NOTE in theory this method can belong to the {@link IndicesRequest} interface because whether a request
          * allowing remote indices has no inherent relationship to whether it is {@link Replaceable} or not.
          * However, we don't have an existing request that is non-replaceable but allows remote indices.
@@ -62,6 +80,90 @@ interface Replaceable extends IndicesRequest {
         default boolean allowsRemoteIndices() {
             return false;
         }
+
+        // TODO probably makes more sense on a service class as opposed to the request itself
+        default <T extends ResponseWithReplacedIndexExpressions> void remoteFanoutErrorHandling(Map<String, T> remoteResults) {}
+    }
+
+    interface CrossProjectSearchCapable extends Replaceable {
+        Logger logger = LogManager.getLogger(CrossProjectSearchCapable.class);
+
+        @Nullable
+        String getProjectRouting();
+
+        @Override
+        default boolean allowsRemoteIndices() {
+            return true;
+        }
+
+        default boolean crossProjectMode() {
+            return getReplacedIndexExpressions() instanceof CrossProjectReplacedIndexExpressions;
+        }
+
+        @Override
+        default <T extends ResponseWithReplacedIndexExpressions> void remoteFanoutErrorHandling(Map<String, T> remoteResults) {
+            logger.info("Checking if we should throw in flat world for [{}]", getReplacedIndexExpressions());
+            // No CPS nothing to do
+            if (false == crossProjectMode()) {
+                logger.info("Skipping because no cross-project expressions found...");
+                return;
+            }
+            if (indicesOptions().allowNoIndices() && indicesOptions().ignoreUnavailable()) {
+                // nothing to do since we're in lenient mode
+                logger.info("Skipping index existence check in lenient mode");
+                return;
+            }
+
+            Map<String, ReplacedIndexExpression> replacedExpressions = getReplacedIndexExpressions().asMap();
+            assert replacedExpressions != null;
+            logger.info("Replaced expressions to check: [{}]", replacedExpressions);
+            for (ReplacedIndexExpression replacedIndexExpression : replacedExpressions.values()) {
+                // TODO need to handle qualified expressions here, too
+                String original = replacedIndexExpression.original();
+                List<ElasticsearchException> exceptions = new ArrayList<>();
+                boolean exists = hasCanonicalExpressionForOrigin(replacedIndexExpression.replacedBy())
+                    && replacedIndexExpression.existsAndVisible();
+                if (exists) {
+                    logger.info("Local cluster has canonical expression for [{}], skipping remote existence check", original);
+                    continue;
+                }
+                if (replacedIndexExpression.authorizationError() != null) {
+                    exceptions.add(replacedIndexExpression.authorizationError());
+                }
+
+                for (var remoteResponse : remoteResults.values()) {
+                    logger.info("Remote response resolved: [{}]", remoteResponse);
+                    Map<String, ReplacedIndexExpression> resolved = remoteResponse.getReplacedIndexExpressions().asMap();
+                    assert resolved != null;
+                    var r = resolved.get(original);
+                    if (r != null && r.existsAndVisible() && resolved.get(original).replacedBy().isEmpty() == false) {
+                        logger.info("Remote cluster has resolved entries for [{}], skipping further remote existence check", original);
+                        exists = true;
+                        break;
+                    } else if (r != null && r.authorizationError() != null) {
+                        assert r.authorized() == false : "we should never get an error if we are authorized";
+                        exceptions.add(resolved.get(original).authorizationError());
+                    }
+                }
+
+                if (false == exists && false == indicesOptions().ignoreUnavailable()) {
+                    if (false == exceptions.isEmpty()) {
+                        // we only ever get exceptions if they are security related
+                        // back and forth on whether a mix or security and non-security (missing indices) exceptions should report
+                        // as 403 or 404
+                        ElasticsearchSecurityException e = new ElasticsearchSecurityException(
+                            "authorization errors while resolving [" + original + "]",
+                            RestStatus.FORBIDDEN
+                        );
+                        exceptions.forEach(e::addSuppressed);
+                        throw e;
+                    } else {
+                        // TODO composite exception based on missing resources
+                        throw new IndexNotFoundException(original);
+                    }
+                }
+            }
+        }
     }
 
     /**
@@ -91,4 +193,5 @@ interface RemoteClusterShardRequest extends IndicesRequest {
          */
         Collection<ShardId> shards();
     }
+
 }

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

@@ -0,0 +1,136 @@
+/*
+ * 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.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.List;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Represents the result of replacing an original index expression with the concrete index resources it resolves to.
+ * Also tracks whether the replacement was authorized, whether the indices exist and are visible, and any
+ * authorization errors encountered during the process.
+ */
+public final class ReplacedIndexExpression implements Writeable {
+    private final String original;
+    private final List<String> replacedBy;
+    private final boolean authorized;
+    private final boolean existsAndVisible;
+    @Nullable
+    private ElasticsearchException authorizationError;
+
+    public ReplacedIndexExpression(StreamInput in) throws IOException {
+        this.original = in.readString();
+        this.replacedBy = in.readCollectionAsList(StreamInput::readString);
+        this.authorized = in.readBoolean();
+        this.existsAndVisible = in.readBoolean();
+        this.authorizationError = ElasticsearchException.readException(in);
+    }
+
+    public ReplacedIndexExpression(
+        String original,
+        List<String> replacedBy,
+        boolean authorized,
+        boolean existsAndVisible,
+        @Nullable ElasticsearchException exception
+    ) {
+        this.original = original;
+        this.replacedBy = replacedBy;
+        this.authorized = authorized;
+        this.existsAndVisible = existsAndVisible;
+        this.authorizationError = exception;
+    }
+
+    public static String[] toIndices(Map<String, ReplacedIndexExpression> replacedExpressions) {
+        return replacedExpressions.values()
+            .stream()
+            .flatMap(indexExpression -> indexExpression.replacedBy().stream())
+            .toArray(String[]::new);
+    }
+
+    public ReplacedIndexExpression(String original, List<String> replacedBy) {
+        this(original, replacedBy, true, true, null);
+    }
+
+    public void setAuthorizationError(ElasticsearchException error) {
+        assert authorized == false : "we should never set an error if we are authorized";
+        this.authorizationError = error;
+    }
+
+    public String original() {
+        return original;
+    }
+
+    public List<String> replacedBy() {
+        return replacedBy;
+    }
+
+    public boolean authorized() {
+        return authorized;
+    }
+
+    public boolean existsAndVisible() {
+        return existsAndVisible;
+    }
+
+    @Nullable
+    public ElasticsearchException authorizationError() {
+        return authorizationError;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (o == null || getClass() != o.getClass()) return false;
+        ReplacedIndexExpression that = (ReplacedIndexExpression) o;
+        return authorized == that.authorized
+            && existsAndVisible == that.existsAndVisible
+            && Objects.equals(original, that.original)
+            && Objects.equals(replacedBy, that.replacedBy)
+            && Objects.equals(authorizationError, that.authorizationError);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(original, replacedBy, authorized, existsAndVisible, authorizationError);
+    }
+
+    @Override
+    public String toString() {
+        return "ReplacedExpression{"
+            + "original='"
+            + original
+            + '\''
+            + ", replacedBy="
+            + replacedBy
+            + ", authorized="
+            + authorized
+            + ", existsAndVisible="
+            + existsAndVisible
+            + ", authorizationError="
+            + authorizationError
+            + '}';
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeString(original);
+        out.writeStringCollection(replacedBy);
+        out.writeBoolean(authorized);
+        out.writeBoolean(existsAndVisible);
+        ElasticsearchException.writeException(authorizationError, out);
+    }
+}