Support point in time cross cluster search (#61827)

This commit integrates point in time into cross cluster search.

Relates #61062
Closes #61790

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

@@ -186,12 +186,14 @@ public static void parseSearchRequest(SearchRequest searchRequest, RestRequest r
         searchRequest.routing(request.param("routing"));
         searchRequest.preference(request.param("preference"));
         searchRequest.indicesOptions(IndicesOptions.fromRequest(request, searchRequest.indicesOptions()));
-        searchRequest.setCcsMinimizeRoundtrips(request.paramAsBoolean("ccs_minimize_roundtrips", searchRequest.isCcsMinimizeRoundtrips()));
 
         checkRestTotalHits(request, searchRequest);
 
         if (searchRequest.pointInTimeBuilder() != null) {
-            preparePointInTime(searchRequest, namedWriteableRegistry);
+            preparePointInTime(searchRequest, request, namedWriteableRegistry);
+        } else {
+            searchRequest.setCcsMinimizeRoundtrips(
+                request.paramAsBoolean("ccs_minimize_roundtrips", searchRequest.isCcsMinimizeRoundtrips()));
         }
     }
 
@@ -308,7 +310,7 @@ private static void parseSearchSource(final SearchSourceBuilder searchSourceBuil
         }
     }
 
-    static void preparePointInTime(SearchRequest request, NamedWriteableRegistry namedWriteableRegistry) {
+    static void preparePointInTime(SearchRequest request, RestRequest restRequest, NamedWriteableRegistry namedWriteableRegistry) {
         assert request.pointInTimeBuilder() != null;
         ActionRequestValidationException validationException = null;
         if (request.indices().length > 0) {
@@ -323,6 +325,11 @@ static void preparePointInTime(SearchRequest request, NamedWriteableRegistry nam
         if (request.preference() != null) {
             validationException = addValidationError("[preference] cannot be used with point in time", validationException);
         }
+        if (restRequest.paramAsBoolean("ccs_minimize_roundtrips", false)) {
+            validationException =
+                addValidationError("[ccs_minimize_roundtrips] cannot be used with point in time", validationException);
+            request.setCcsMinimizeRoundtrips(false);
+        }
         ExceptionsHelper.reThrowIfNotNull(validationException);
 
         final IndicesOptions indicesOptions = request.indicesOptions();

server/src/test/java/org/elasticsearch/action/search/TransportSearchActionTests.java

@@ -208,7 +208,6 @@ public void testProcessRemoteShards() {
             null)) {
             RemoteClusterService service = transportService.getRemoteClusterService();
             assertFalse(service.isCrossClusterSearchEnabled());
-            List<SearchShardIterator> iteratorList = new ArrayList<>();
             Map<String, ClusterSearchShardsResponse> searchShardsResponseMap = new HashMap<>();
             DiscoveryNode[] nodes = new DiscoveryNode[] {
                 new DiscoveryNode("node1", buildNewFakeTransportAddress(), Version.CURRENT),
@@ -246,9 +245,9 @@ public void testProcessRemoteShards() {
                 new OriginalIndices(new String[]{"fo*", "ba*"}, SearchRequest.DEFAULT_INDICES_OPTIONS));
             remoteIndicesByCluster.put("test_cluster_2",
                 new OriginalIndices(new String[]{"x*"}, SearchRequest.DEFAULT_INDICES_OPTIONS));
-            Map<String, AliasFilter> remoteAliases = new HashMap<>();
-            TransportSearchAction.processRemoteShards(searchShardsResponseMap, remoteIndicesByCluster, iteratorList,
-                remoteAliases);
+            Map<String, AliasFilter> remoteAliases = TransportSearchAction.getRemoteAliasFilters(searchShardsResponseMap);
+            List<SearchShardIterator> iteratorList =
+                TransportSearchAction.getRemoteShardsIterator(searchShardsResponseMap, remoteIndicesByCluster, remoteAliases);
             assertEquals(4, iteratorList.size());
             for (SearchShardIterator iterator : iteratorList) {
                 if (iterator.shardId().getIndexName().endsWith("foo")) {

test/framework/src/main/java/org/elasticsearch/test/AbstractMultiClustersTestCase.java

@@ -0,0 +1,199 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.test;
+
+import org.elasticsearch.action.admin.cluster.remote.RemoteInfoAction;
+import org.elasticsearch.action.admin.cluster.remote.RemoteInfoRequest;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.common.network.NetworkModule;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.core.internal.io.IOUtils;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.test.transport.MockTransportService;
+import org.elasticsearch.transport.RemoteClusterAware;
+import org.elasticsearch.transport.RemoteConnectionInfo;
+import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.transport.nio.MockNioTransportPlugin;
+import org.junit.AfterClass;
+import org.junit.Before;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import static org.elasticsearch.discovery.DiscoveryModule.DISCOVERY_SEED_PROVIDERS_SETTING;
+import static org.elasticsearch.discovery.SettingsBasedSeedHostsProvider.DISCOVERY_SEED_HOSTS_SETTING;
+import static org.hamcrest.Matchers.hasKey;
+import static org.hamcrest.Matchers.hasSize;
+
+public abstract class AbstractMultiClustersTestCase extends ESTestCase {
+    public static final String LOCAL_CLUSTER = RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY;
+
+    private static volatile ClusterGroup clusterGroup;
+
+    protected Collection<String> remoteClusterAlias() {
+        return randomSubsetOf(Arrays.asList("cluster-a", "cluster-b"));
+    }
+
+    protected Collection<Class<? extends Plugin>> nodePlugins(String clusterAlias) {
+        return Collections.emptyList();
+    }
+
+    protected final Client client() {
+        return client(LOCAL_CLUSTER);
+    }
+
+    protected final Client client(String clusterAlias) {
+        return cluster(clusterAlias).client();
+    }
+
+    protected final InternalTestCluster cluster(String clusterAlias) {
+        return clusterGroup.getCluster(clusterAlias);
+    }
+
+    protected final Map<String, InternalTestCluster> clusters() {
+        return Collections.unmodifiableMap(clusterGroup.clusters);
+    }
+
+    protected boolean reuseClusters() {
+        return true;
+    }
+
+    @Before
+    public final void startClusters() throws Exception {
+        if (clusterGroup != null && reuseClusters()) {
+            return;
+        }
+        stopClusters();
+        final Map<String, InternalTestCluster> clusters = new HashMap<>();
+        final List<String> clusterAliases = new ArrayList<>(remoteClusterAlias());
+        clusterAliases.add(LOCAL_CLUSTER);
+        for (String clusterAlias : clusterAliases) {
+            final String clusterName = clusterAlias.equals(LOCAL_CLUSTER) ? "main-cluster" : clusterAlias;
+            final int numberOfNodes = randomIntBetween(1, 3);
+            final List<Class<? extends Plugin>> mockPlugins =
+               Arrays.asList(MockHttpTransport.TestPlugin.class, MockTransportService.TestPlugin.class, MockNioTransportPlugin.class);
+            final Collection<Class<? extends Plugin>> nodePlugins = nodePlugins(clusterAlias);
+            final Settings nodeSettings = Settings.EMPTY;
+            final NodeConfigurationSource nodeConfigurationSource = nodeConfigurationSource(nodeSettings, nodePlugins);
+            final InternalTestCluster cluster = new InternalTestCluster(randomLong(), createTempDir(), true, true, numberOfNodes,
+                numberOfNodes, clusterName, nodeConfigurationSource, 0, clusterName + "-", mockPlugins, Function.identity());
+            cluster.beforeTest(random(), 0);
+            clusters.put(clusterAlias, cluster);
+        }
+        clusterGroup = new ClusterGroup(clusters);
+        configureRemoteClusters();
+    }
+
+    @AfterClass
+    public static void stopClusters() throws IOException {
+        IOUtils.close(clusterGroup);
+        clusterGroup = null;
+    }
+
+    private void configureRemoteClusters() throws Exception {
+        Map<String, List<String>> seedNodes = new HashMap<>();
+        for (String clusterAlias : clusterGroup.clusterAliases()) {
+            if (clusterAlias.equals(LOCAL_CLUSTER) == false) {
+                final InternalTestCluster cluster = clusterGroup.getCluster(clusterAlias);
+                final String[] allNodes = cluster.getNodeNames();
+                final List<String> selectedNodes = randomSubsetOf(randomIntBetween(1, Math.min(3, allNodes.length)), allNodes);
+                seedNodes.put(clusterAlias, selectedNodes);
+            }
+        }
+        if (seedNodes.isEmpty()) {
+            return;
+        }
+        Settings.Builder settings = Settings.builder();
+        for (Map.Entry<String, List<String>> entry : seedNodes.entrySet()) {
+            final String clusterAlias = entry.getKey();
+            final String seeds = entry.getValue().stream()
+                .map(node -> cluster(clusterAlias).getInstance(TransportService.class, node).boundAddress().publishAddress().toString())
+                .collect(Collectors.joining(","));
+            settings.put("cluster.remote." + clusterAlias + ".seeds", seeds);
+        }
+        client().admin().cluster().prepareUpdateSettings().setPersistentSettings(settings).get();
+        assertBusy(() -> {
+            List<RemoteConnectionInfo> remoteConnectionInfos = client()
+                .execute(RemoteInfoAction.INSTANCE, new RemoteInfoRequest()).actionGet().getInfos()
+                .stream().filter(RemoteConnectionInfo::isConnected)
+                .collect(Collectors.toList());
+            final long totalConnections = seedNodes.values().stream().map(List::size).count();
+            assertThat(remoteConnectionInfos, hasSize(Math.toIntExact(totalConnections)));
+        });
+    }
+
+    static class ClusterGroup implements Closeable {
+        private final Map<String, InternalTestCluster> clusters;
+
+        ClusterGroup(Map<String, InternalTestCluster> clusters) {
+            this.clusters = Collections.unmodifiableMap(clusters);
+        }
+
+        InternalTestCluster getCluster(String clusterAlias) {
+            assertThat(clusters, hasKey(clusterAlias));
+            return clusters.get(clusterAlias);
+        }
+
+        Set<String> clusterAliases() {
+            return clusters.keySet();
+        }
+
+        @Override
+        public void close() throws IOException {
+            IOUtils.close(clusters.values());
+        }
+    }
+
+    static NodeConfigurationSource nodeConfigurationSource(Settings nodeSettings, Collection<Class<? extends Plugin>> nodePlugins) {
+        final Settings.Builder builder = Settings.builder();
+        builder.putList(DISCOVERY_SEED_HOSTS_SETTING.getKey()); // empty list disables a port scan for other nodes
+        builder.putList(DISCOVERY_SEED_PROVIDERS_SETTING.getKey(), "file");
+        builder.put(NetworkModule.TRANSPORT_TYPE_KEY, getTestTransportType());
+        builder.put(nodeSettings);
+
+        return new NodeConfigurationSource() {
+            @Override
+            public Settings nodeSettings(int nodeOrdinal) {
+                return builder.build();
+            }
+
+            @Override
+            public Path nodeConfigPath(int nodeOrdinal) {
+                return null;
+            }
+
+            @Override
+            public Collection<Class<? extends Plugin>> nodePlugins() {
+                return nodePlugins;
+            }
+        };
+    }
+}

x-pack/plugin/core/src/internalClusterTest/java/org/elasticsearch/xpack/core/search/CCSPointInTimeIT.java

@@ -0,0 +1,107 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.core.search;
+
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.index.query.MatchAllQueryBuilder;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.test.AbstractMultiClustersTestCase;
+import org.elasticsearch.xpack.core.LocalStateCompositeXPackPlugin;
+import org.elasticsearch.xpack.core.search.action.ClosePointInTimeAction;
+import org.elasticsearch.xpack.core.search.action.ClosePointInTimeRequest;
+import org.elasticsearch.xpack.core.search.action.OpenPointInTimeAction;
+import org.elasticsearch.xpack.core.search.action.OpenPointInTimeRequest;
+import org.elasticsearch.xpack.core.search.action.OpenPointInTimeResponse;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoFailures;
+
+public class CCSPointInTimeIT extends AbstractMultiClustersTestCase {
+
+    @Override
+    protected Collection<String> remoteClusterAlias() {
+        return Collections.singletonList("remote_cluster");
+    }
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins(String clusterAlias) {
+        final List<Class<? extends Plugin>> plugins = new ArrayList<>(super.nodePlugins(clusterAlias));
+        plugins.add(LocalStateCompositeXPackPlugin.class);
+        return plugins;
+    }
+
+    void indexDocs(Client client, String index, int numDocs) {
+        for (int i = 0; i < numDocs; i++) {
+            String id = Integer.toString(i);
+            client.prepareIndex(index, "_doc").setId(id).setSource("value", i).get();
+        }
+        client.admin().indices().prepareRefresh(index).get();
+    }
+
+    public void testBasic() {
+        final Client localClient = client(LOCAL_CLUSTER);
+        final Client remoteClient = client("remote_cluster");
+        int localNumDocs = randomIntBetween(10, 50);
+        assertAcked(localClient.admin().indices().prepareCreate("local_test"));
+        indexDocs(localClient, "local_test", localNumDocs);
+
+        int remoteNumDocs = randomIntBetween(10, 50);
+        assertAcked(remoteClient.admin().indices().prepareCreate("remote_test"));
+        indexDocs(remoteClient, "remote_test", remoteNumDocs);
+        boolean includeLocalIndex = randomBoolean();
+        List<String> indices = new ArrayList<>();
+        if (includeLocalIndex) {
+            indices.add( randomFrom("*", "local_*", "local_test"));
+        }
+        indices.add(randomFrom("*:*", "remote_cluster:*", "remote_cluster:remote_test"));
+        String pitId = openPointInTime(indices.toArray(new String[0]), TimeValue.timeValueMinutes(2));
+        try {
+            if (randomBoolean()) {
+                localClient.prepareIndex("local_test", "_doc").setId("local_new").setSource().get();
+                localClient.admin().indices().prepareRefresh().get();
+            }
+            if (randomBoolean()) {
+                remoteClient.prepareIndex("remote_test", "_doc").setId("remote_new").setSource().get();
+                remoteClient.admin().indices().prepareRefresh().get();
+            }
+            SearchResponse resp = localClient.prepareSearch()
+                .setPreference(null)
+                .setQuery(new MatchAllQueryBuilder())
+                .setSearchContext(pitId, TimeValue.timeValueMinutes(2))
+                .setSize(1000)
+                .get();
+            assertNoFailures(resp);
+            assertHitCount(resp, (includeLocalIndex ? localNumDocs : 0) + remoteNumDocs);
+        } finally {
+            closePointInTime(pitId);
+        }
+    }
+
+    private String openPointInTime(String[] indices, TimeValue keepAlive) {
+        OpenPointInTimeRequest request = new OpenPointInTimeRequest(
+            indices,
+            OpenPointInTimeRequest.DEFAULT_INDICES_OPTIONS,
+            keepAlive,
+            null,
+            null
+        );
+        final OpenPointInTimeResponse response = client().execute(OpenPointInTimeAction.INSTANCE, request).actionGet();
+        return response.getSearchContextId();
+    }
+
+    private void closePointInTime(String readerId) {
+        client().execute(ClosePointInTimeAction.INSTANCE, new ClosePointInTimeRequest(readerId)).actionGet();
+    }
+}

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/search/action/OpenPointInTimeResponse.java

@@ -14,14 +14,15 @@
 import org.elasticsearch.common.xcontent.XContentBuilder;
 
 import java.io.IOException;
+import java.util.Objects;
 
 public final class OpenPointInTimeResponse extends ActionResponse implements ToXContentObject {
     private static final ParseField ID = new ParseField("id");
 
     private final String searchContextId;
 
     public OpenPointInTimeResponse(String searchContextId) {
-        this.searchContextId = searchContextId;
+        this.searchContextId = Objects.requireNonNull(searchContextId);
     }
 
     public OpenPointInTimeResponse(StreamInput in) throws IOException {