Allow users to run the painless execute API on a remote cluster shard (#97335)

Added a clusterAlias to the Painless execute Request object, so that index
expressions in the request of the form "myremote:myindex" will be parsed to
set clusterAlias to "myremote" and the index to "myindex".

If clusterAlias is null, then it is executed against a shard on the local cluster, as before.
If clusterAlias is non-null, then the SingleShardTransportAction is sent to the remote cluster,
where it will run the full request (doing remote coordination). Note that the new clusterAlias 
field is not Writeable so that when it is sent to the remote cluster it will only see the index
name, not the clusterAlias (which it wouldn't know how to handle correctly).

Added PainlessExecuteIT test that tests cross-cluster calls

Updated painless-execute-script end user docs to indicate support for cross-cluster executions

docs/painless/painless-guide/painless-execute-script.asciidoc

@@ -118,6 +118,11 @@ Document that's temporarily indexed in-memory and accessible from the script.
 
 `index`:: (Required, string)
 Index containing a mapping that's compatible with the indexed document.
+You may specify a remote index by prefixing the index with the remote cluster
+alias. For example, `remote1:my_index` indicates that you want to execute
+the painless script against the "my_index" index on the "remote1" cluster. This
+request will be forwarded to the "remote1" cluster if you have
+{ref}/remote-clusters-connect.html[configured a connection] to that remote cluster.
 ====
 
 `params`:: (`Map`, read-only)

docs/reference/search/search-your-data/search-across-clusters.asciidoc

@@ -17,6 +17,7 @@ The following APIs support {ccs}:
 * <<search-template,Search template>>
 * <<multi-search-template,Multi search template>>
 * <<search-field-caps,Field capabilities>>
+* {painless}/painless-execute-api.html[Painless execute API]
 * experimental:[] <<eql-search-api,EQL search>>
 * experimental:[] <<sql-search-api,SQL search>>
 * experimental:[] <<search-vector-tile-api,Vector tile search>>

modules/lang-painless/build.gradle

@@ -11,6 +11,7 @@ import org.elasticsearch.gradle.testclusters.DefaultTestClustersTask;
 apply plugin: 'elasticsearch.validate-rest-spec'
 apply plugin: 'elasticsearch.legacy-yaml-rest-test'
 apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
+apply plugin: 'elasticsearch.internal-cluster-test'
 
 esplugin {
     description 'An easy, safe and fast scripting language for Elasticsearch'

modules/lang-painless/src/internalClusterTest/java/org/elasticsearch/painless/action/CrossClusterPainlessExecuteIT.java

@@ -0,0 +1,195 @@
+/*
+ * 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 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 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.painless.action;
+
+import org.elasticsearch.action.ActionFuture;
+import org.elasticsearch.client.internal.Client;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.core.Strings;
+import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.index.query.MatchAllQueryBuilder;
+import org.elasticsearch.index.query.QueryBuilder;
+import org.elasticsearch.painless.PainlessPlugin;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.script.FilterScript;
+import org.elasticsearch.script.Script;
+import org.elasticsearch.script.ScriptContext;
+import org.elasticsearch.script.ScriptType;
+import org.elasticsearch.test.AbstractMultiClustersTestCase;
+import org.elasticsearch.test.InternalTestCluster;
+import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.XContentType;
+import org.hamcrest.Matchers;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
+
+/**
+ * Tests the PainlessExecuteAction against a local cluster with one remote cluster configured.
+ * Execute action tests are run against both the local cluster and the remote cluster.
+ */
+public class CrossClusterPainlessExecuteIT extends AbstractMultiClustersTestCase {
+
+    private static final String REMOTE_CLUSTER = "cluster_a";
+    private static final String LOCAL_INDEX = "local_idx";
+    private static final String REMOTE_INDEX = "remote_idx";
+    private static final String KEYWORD_FIELD = "my_field";
+
+    @Override
+    protected Collection<String> remoteClusterAlias() {
+        return List.of(REMOTE_CLUSTER);
+    }
+
+    @Override
+    protected boolean reuseClusters() {
+        return false;
+    }
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins(String clusterAlias) {
+        List<Class<? extends Plugin>> plugs = Arrays.asList(PainlessPlugin.class);
+        return Stream.concat(super.nodePlugins(clusterAlias).stream(), plugs.stream()).collect(Collectors.toList());
+    }
+
+    public void testPainlessExecuteAgainstLocalCluster() throws Exception {
+        setupTwoClusters();
+
+        Script script = new Script(
+            ScriptType.INLINE,
+            Script.DEFAULT_SCRIPT_LANG,
+            Strings.format("doc['%s'].value.length() <= params.max_length", KEYWORD_FIELD),
+            Map.of("max_length", 4)
+        );
+        ScriptContext<?> context = FilterScript.CONTEXT;
+
+        PainlessExecuteAction.Request.ContextSetup contextSetup = createContextSetup(LOCAL_INDEX);
+        PainlessExecuteAction.Request request = new PainlessExecuteAction.Request(script, context.name, contextSetup);
+
+        ActionFuture<PainlessExecuteAction.Response> actionFuture = client(LOCAL_CLUSTER).admin()
+            .cluster()
+            .execute(PainlessExecuteAction.INSTANCE, request);
+
+        PainlessExecuteAction.Response response = actionFuture.actionGet();
+        Object result = response.getResult();
+        assertThat(result, Matchers.instanceOf(Boolean.class));
+        assertTrue((Boolean) result);
+    }
+
+    /**
+     * Query the local cluster to run the execute actions against the 'cluster_a:remote_idx' index.
+     * There is no local index with the REMOTE_INDEX name, so it has to do a cross-cluster action for this to work
+     */
+    public void testPainlessExecuteAsCrossClusterAction() throws Exception {
+        setupTwoClusters();
+
+        Script script = new Script(
+            ScriptType.INLINE,
+            Script.DEFAULT_SCRIPT_LANG,
+            Strings.format("doc['%s'].value.length() <= params.max_length", KEYWORD_FIELD),
+            Map.of("max_length", 4)
+        );
+        ScriptContext<?> context = FilterScript.CONTEXT;
+
+        PainlessExecuteAction.Request.ContextSetup contextSetup = createContextSetup(REMOTE_CLUSTER + ":" + REMOTE_INDEX);
+        PainlessExecuteAction.Request request = new PainlessExecuteAction.Request(script, context.name, contextSetup);
+
+        ActionFuture<PainlessExecuteAction.Response> actionFuture = client(LOCAL_CLUSTER).admin()
+            .cluster()
+            .execute(PainlessExecuteAction.INSTANCE, request);
+
+        PainlessExecuteAction.Response response = actionFuture.actionGet();
+        Object result = response.getResult();
+        assertThat(result, Matchers.instanceOf(Boolean.class));
+        assertTrue((Boolean) result);
+    }
+
+    private static PainlessExecuteAction.Request.ContextSetup createContextSetup(String index) {
+        QueryBuilder query = new MatchAllQueryBuilder();
+        BytesReference doc;
+        XContentType xContentType = XContentType.JSON.canonical();
+        try {
+            XContentBuilder xContentBuilder = XContentBuilder.builder(xContentType.xContent());
+            xContentBuilder.startObject();
+            xContentBuilder.field(KEYWORD_FIELD, "four");
+            xContentBuilder.endObject();
+            doc = BytesReference.bytes(xContentBuilder);
+        } catch (IOException e) {
+            throw new UncheckedIOException(e);
+        }
+
+        PainlessExecuteAction.Request.ContextSetup contextSetup = new PainlessExecuteAction.Request.ContextSetup(index, doc, query);
+        contextSetup.setXContentType(XContentType.JSON);
+        return contextSetup;
+    }
+
+    private void setupTwoClusters() throws Exception {
+        assertAcked(client(LOCAL_CLUSTER).admin().indices().prepareCreate(LOCAL_INDEX).setMapping(KEYWORD_FIELD, "type=keyword"));
+        indexDocs(client(LOCAL_CLUSTER), LOCAL_INDEX);
+        final InternalTestCluster remoteCluster = cluster(REMOTE_CLUSTER);
+        remoteCluster.ensureAtLeastNumDataNodes(1);
+        final Settings.Builder allocationFilter = Settings.builder();
+        if (randomBoolean()) {
+            remoteCluster.ensureAtLeastNumDataNodes(3);
+            List<String> remoteDataNodes = remoteCluster.clusterService()
+                .state()
+                .nodes()
+                .stream()
+                .filter(DiscoveryNode::canContainData)
+                .map(DiscoveryNode::getName)
+                .toList();
+            assertThat(remoteDataNodes.size(), Matchers.greaterThanOrEqualTo(3));
+            List<String> seedNodes = randomSubsetOf(between(1, remoteDataNodes.size() - 1), remoteDataNodes);
+            disconnectFromRemoteClusters();
+            configureRemoteCluster(REMOTE_CLUSTER, seedNodes);
+            if (randomBoolean()) {
+                // Using proxy connections
+                allocationFilter.put("index.routing.allocation.exclude._name", String.join(",", seedNodes));
+            } else {
+                allocationFilter.put("index.routing.allocation.include._name", String.join(",", seedNodes));
+            }
+        }
+        assertAcked(
+            client(REMOTE_CLUSTER).admin()
+                .indices()
+                .prepareCreate(REMOTE_INDEX)
+                .setMapping(KEYWORD_FIELD, "type=keyword")
+                .setSettings(Settings.builder().put(allocationFilter.build()).put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0))
+        );
+        assertFalse(
+            client(REMOTE_CLUSTER).admin()
+                .cluster()
+                .prepareHealth(REMOTE_INDEX)
+                .setWaitForYellowStatus()
+                .setTimeout(TimeValue.timeValueSeconds(10))
+                .get()
+                .isTimedOut()
+        );
+        indexDocs(client(REMOTE_CLUSTER), REMOTE_INDEX);
+    }
+
+    private int indexDocs(Client client, String index) {
+        int numDocs = between(1, 10);
+        for (int i = 0; i < numDocs; i++) {
+            client.prepareIndex(index).setSource(KEYWORD_FIELD, "my_value").get();
+        }
+        client.admin().indices().prepareRefresh(index).get();
+        return numDocs;
+    }
+}

modules/lang-painless/src/main/java/org/elasticsearch/painless/action/PainlessExecuteAction.java

@@ -20,6 +20,7 @@ import org.apache.lucene.search.Scorer;
 import org.apache.lucene.search.Weight;
 import org.apache.lucene.store.ByteBuffersDirectory;
 import org.apache.lucene.store.Directory;
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.action.ActionType;
@@ -27,6 +28,7 @@ import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.action.support.single.shard.SingleShardRequest;
 import org.elasticsearch.action.support.single.shard.TransportSingleShardAction;
+import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.client.internal.node.NodeClient;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.block.ClusterBlockException;
@@ -44,6 +46,9 @@ import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.network.NetworkAddress;
 import org.elasticsearch.common.xcontent.LoggingDeprecationHandler;
 import org.elasticsearch.common.xcontent.XContentHelper;
+import org.elasticsearch.core.Nullable;
+import org.elasticsearch.core.Strings;
+import org.elasticsearch.core.Tuple;
 import org.elasticsearch.geometry.Point;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.IndexService;
@@ -80,6 +85,7 @@ import org.elasticsearch.script.ScriptService;
 import org.elasticsearch.script.ScriptType;
 import org.elasticsearch.script.StringFieldScript;
 import org.elasticsearch.search.lookup.SearchLookup;
+import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xcontent.ConstructingObjectParser;
@@ -175,6 +181,8 @@ public class PainlessExecuteAction extends ActionType<PainlessExecuteAction.Resp
                 );
             }
 
+            @Nullable // null means local cluster
+            private final transient String clusterAlias;  // this field is not Writeable, as it is needed only on the initial receiving node
             private final String index;
             private final BytesReference document;
             private final QueryBuilder query;
@@ -188,13 +196,16 @@ public class PainlessExecuteAction extends ActionType<PainlessExecuteAction.Resp
             }
 
             ContextSetup(String index, BytesReference document, QueryBuilder query) {
-                this.index = index;
+                Tuple<String, String> clusterAliasAndIndex = parseClusterAliasAndIndex(index);
+                this.clusterAlias = clusterAliasAndIndex.v1();
+                this.index = clusterAliasAndIndex.v2();
                 this.document = document;
                 this.query = query;
             }
 
             ContextSetup(StreamInput in) throws IOException {
-                index = in.readOptionalString();
+                this.clusterAlias = null;
+                this.index = in.readOptionalString();
                 document = in.readOptionalBytesReference();
                 String optionalXContentType = in.readOptionalString();
                 if (optionalXContentType != null) {
@@ -203,6 +214,47 @@ public class PainlessExecuteAction extends ActionType<PainlessExecuteAction.Resp
                 query = in.readOptionalNamedWriteable(QueryBuilder.class);
             }
 
+            /**
+             * @param indexExpression should be of the form "index" or "cluster:index". Wildcards are OK.
+             * @return Tuple where first entry is clusterAlias, which will be null if not in the indexExpression
+             *         and second entry is the index name
+             *         Tuple(null, null) will be returned if indexExpression is null
+             * @throws IllegalArgumentException if the indexExpression starts or ends with the REMOTE_CLUSTER_INDEX_SEPARATOR (":")
+             *         (ignoring whitespace)
+             */
+            static Tuple<String, String> parseClusterAliasAndIndex(String indexExpression) {
+                if (indexExpression == null) {
+                    return new Tuple<>(null, null);
+                }
+                String trimmed = indexExpression.trim();
+                if (trimmed.startsWith(":") || trimmed.endsWith(":")) {
+                    throw new IllegalArgumentException(
+                        "Unable to parse one single valid index name from the provided index: [" + indexExpression + "]"
+                    );
+                }
+
+                // The parser here needs to ensure that the indexExpression is not of the form "remote1:blogs,remote2:blogs"
+                // because (1) only a single index is allowed for Painless Execute and
+                // (2) if this method returns Tuple("remote1", "blogs,remote2:blogs") that will not fail with "index not found".
+                // Instead, it will fail with the inaccurate and confusing error message:
+                // "Cross-cluster calls are not supported in this context but remote indices were requested: [blogs,remote1:blogs]"
+                // which comes later out of the IndexNameExpressionResolver pathway this code uses.
+                String[] parts = indexExpression.split(":", 2);
+                if (parts.length == 1) {
+                    return new Tuple<>(null, parts[0]);
+                } else if (parts.length == 2 && parts[1].contains(":") == false) {
+                    return new Tuple<>(parts[0], parts[1]);
+                } else {
+                    throw new IllegalArgumentException(
+                        "Unable to parse one single valid index name from the provided index: [" + indexExpression + "]"
+                    );
+                }
+            }
+
+            public String getClusterAlias() {
+                return clusterAlias;
+            }
+
             public String getIndex() {
                 return index;
             }
@@ -241,6 +293,8 @@ public class PainlessExecuteAction extends ActionType<PainlessExecuteAction.Resp
 
             @Override
             public void writeTo(StreamOutput out) throws IOException {
+                // clusterAlias is not included as only the original coordinator needs to see it
+                // if forwarded to a remote cluster, the remote cluster will execute it locally
                 out.writeOptionalString(index);
                 out.writeOptionalBytesReference(document);
                 out.writeOptionalString(xContentType != null ? xContentType.mediaTypeWithoutParameters() : null);
@@ -249,17 +303,14 @@ public class PainlessExecuteAction extends ActionType<PainlessExecuteAction.Resp
 
             @Override
             public String toString() {
-                return "ContextSetup{"
-                    + ", index='"
-                    + index
-                    + '\''
-                    + ", document="
-                    + document
-                    + ", query="
-                    + query
-                    + ", xContentType="
-                    + xContentType
-                    + '}';
+                return Strings.format(
+                    "ContextSetup{index=%s, cluster=%s, document=%s, query=%s, xContentType=%s}",
+                    index,
+                    clusterAlias,
+                    document,
+                    query,
+                    xContentType
+                );
             }
 
             @Override
@@ -469,6 +520,18 @@ public class PainlessExecuteAction extends ActionType<PainlessExecuteAction.Resp
             this.indicesServices = indicesServices;
         }
 
+        @Override
+        protected void doExecute(Task task, Request request, ActionListener<Response> listener) {
+            if (request.getContextSetup() == null || request.getContextSetup().getClusterAlias() == null) {
+                super.doExecute(task, request, listener);
+            } else {
+                // forward to remote cluster
+                String clusterAlias = request.getContextSetup().getClusterAlias();
+                Client remoteClusterClient = transportService.getRemoteClusterService().getRemoteClusterClient(threadPool, clusterAlias);
+                remoteClusterClient.admin().cluster().execute(PainlessExecuteAction.INSTANCE, request, listener);
+            }
+        }
+
         @Override
         protected Writeable.Reader<Response> getResponseReader() {
             return Response::new;

modules/lang-painless/src/test/java/org/elasticsearch/painless/action/PainlessExecuteApiTests.java

@@ -9,6 +9,7 @@ package org.elasticsearch.painless.action;
 
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.core.Tuple;
 import org.elasticsearch.index.IndexService;
 import org.elasticsearch.index.query.MatchAllQueryBuilder;
 import org.elasticsearch.index.query.MatchQueryBuilder;
@@ -396,4 +397,77 @@ public class PainlessExecuteApiTests extends ESSingleNodeTestCase {
         assertEquals(2, Integer.parseInt((String) response.getResult()));
     }
 
+    /**
+     * When an index expression with a remote cluster name is passed into Request.ContextSetup, it
+     * is parsed into separate fields - clusterAlias and index.
+     * The other tests in this suite test without a clusterAlias prefix.
+     * This test ensures that innerShardOperation works the same with one present, since the clusterAlias
+     * field is only needed by the initial coordinator of the action to determine where to run the
+     * action (which is not part of the tests in this suite).
+     */
+    public void testFilterExecutionContextWorksWithRemoteClusterPrefix() throws IOException {
+        ScriptService scriptService = getInstanceFromNode(ScriptService.class);
+        String indexName = "index";
+        IndexService indexService = createIndex(indexName, Settings.EMPTY, "doc", "field", "type=long");
+
+        String indexNameWithClusterAlias = "remote1:" + indexName;
+        Request.ContextSetup contextSetup = new Request.ContextSetup(indexNameWithClusterAlias, new BytesArray("{\"field\": 3}"), null);
+        contextSetup.setXContentType(XContentType.JSON);
+        Request request = new Request(new Script("doc['field'].value >= 3"), "filter", contextSetup);
+        Response response = innerShardOperation(request, scriptService, indexService);
+        assertThat(response.getResult(), equalTo(true));
+
+        contextSetup = new Request.ContextSetup(indexNameWithClusterAlias, new BytesArray("{\"field\": 3}"), null);
+        contextSetup.setXContentType(XContentType.JSON);
+        request = new Request(
+            new Script(ScriptType.INLINE, "painless", "doc['field'].value >= params.max", singletonMap("max", 3)),
+            "filter",
+            contextSetup
+        );
+        response = innerShardOperation(request, scriptService, indexService);
+        assertThat(response.getResult(), equalTo(true));
+
+        contextSetup = new Request.ContextSetup(indexNameWithClusterAlias, new BytesArray("{\"field\": 2}"), null);
+        contextSetup.setXContentType(XContentType.JSON);
+        request = new Request(
+            new Script(ScriptType.INLINE, "painless", "doc['field'].value >= params.max", singletonMap("max", 3)),
+            "filter",
+            contextSetup
+        );
+        response = innerShardOperation(request, scriptService, indexService);
+        assertThat(response.getResult(), equalTo(false));
+    }
+
+    public void testParseClusterAliasAndIndex() {
+        record ValidTestCase(String input, Tuple<String, String> output) {}
+
+        ValidTestCase[] cases = new ValidTestCase[] {
+            // valid index expressions
+            new ValidTestCase("remote1:foo", new Tuple<>("remote1", "foo")),
+            new ValidTestCase("foo", new Tuple<>(null, "foo")),
+            new ValidTestCase("foo,bar", new Tuple<>(null, "foo,bar")), // this method only checks for invalid ":"
+            new ValidTestCase("", new Tuple<>(null, "")),
+            new ValidTestCase(null, new Tuple<>(null, null)) };
+
+        for (ValidTestCase testCase : cases) {
+            Tuple<String, String> output = Request.ContextSetup.parseClusterAliasAndIndex(testCase.input);
+            assertEquals(testCase.output(), output);
+        }
+
+        expectThrows(IllegalArgumentException.class, () -> Request.ContextSetup.parseClusterAliasAndIndex("remote1::foo"));
+        expectThrows(IllegalArgumentException.class, () -> Request.ContextSetup.parseClusterAliasAndIndex("remote1:foo:"));
+        expectThrows(IllegalArgumentException.class, () -> Request.ContextSetup.parseClusterAliasAndIndex(" :remote1:foo"));
+        expectThrows(IllegalArgumentException.class, () -> Request.ContextSetup.parseClusterAliasAndIndex("remote1:foo: "));
+        expectThrows(IllegalArgumentException.class, () -> Request.ContextSetup.parseClusterAliasAndIndex("remote1::::"));
+        expectThrows(IllegalArgumentException.class, () -> Request.ContextSetup.parseClusterAliasAndIndex("::"));
+        expectThrows(IllegalArgumentException.class, () -> Request.ContextSetup.parseClusterAliasAndIndex(":x:"));
+        expectThrows(IllegalArgumentException.class, () -> Request.ContextSetup.parseClusterAliasAndIndex(" : : "));
+        expectThrows(IllegalArgumentException.class, () -> Request.ContextSetup.parseClusterAliasAndIndex(":blogs:"));
+        expectThrows(IllegalArgumentException.class, () -> Request.ContextSetup.parseClusterAliasAndIndex(":blogs"));
+        expectThrows(IllegalArgumentException.class, () -> Request.ContextSetup.parseClusterAliasAndIndex(" :blogs"));
+        expectThrows(IllegalArgumentException.class, () -> Request.ContextSetup.parseClusterAliasAndIndex("blogs:"));
+        expectThrows(IllegalArgumentException.class, () -> Request.ContextSetup.parseClusterAliasAndIndex("blogs:  "));
+        expectThrows(IllegalArgumentException.class, () -> Request.ContextSetup.parseClusterAliasAndIndex("remote1:foo,remote2:bar"));
+        expectThrows(IllegalArgumentException.class, () -> Request.ContextSetup.parseClusterAliasAndIndex("a:b,c:d,e:f"));
+    }
 }