Support point in time cross cluster search (#61827)

This commit integrates point in time into cross cluster search.

Relates #61062
Closes #61790

Author: dnhatn

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

@@ -168,16 +168,17 @@ public static void parseSearchRequest(SearchRequest searchRequest, RestRequest r
         if (scroll != null) {
             searchRequest.scroll(new Scroll(parseTimeValue(scroll, null, "scroll")));
         }
-
         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()));
         }
     }
 
@@ -293,7 +294,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) {
@@ -308,6 +309,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/main/java/org/elasticsearch/rest/action/search/RestSearchAction.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")) {

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

@@ -89,7 +89,6 @@ public void setupSuiteScopeCluster() throws InterruptedException {
         indexRandom(true, true, reqs);
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/61790")
     public void testMaxMinAggregation() throws Exception {
         int step = numShards > 2 ? randomIntBetween(2, numShards) : 2;
         int numFailures = randomBoolean() ? randomIntBetween(0, numShards) : 0;
@@ -134,7 +133,6 @@ public void testMaxMinAggregation() throws Exception {
         }
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/61790")
     public void testTermsAggregation() throws Exception {
         int step = numShards > 2 ? randomIntBetween(2, numShards) : 2;
         int numFailures = randomBoolean() ? randomIntBetween(0, numShards) : 0;

x-pack/plugin/async-search/src/internalClusterTest/java/org/elasticsearch/xpack/search/AsyncSearchActionIT.java

@@ -0,0 +1,106 @@
+/*
+ * 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.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 List.of("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).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").setId("local_new").setSource().get();
+                localClient.admin().indices().prepareRefresh().get();
+            }
+            if (randomBoolean()) {
+                remoteClient.prepareIndex("remote_test").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/internalClusterTest/java/org/elasticsearch/xpack/core/search/CCSPointInTimeIT.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 {

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

@@ -72,6 +72,7 @@ protected void doExecute(Task task, OpenPointInTimeRequest request, ActionListen
             .preference(request.preference())
             .routing(request.routing())
             .allowPartialSearchResults(false);
+        searchRequest.setCcsMinimizeRoundtrips(false);
         transportSearchAction.executeRequest(
             task,
             searchRequest,
@@ -91,7 +92,10 @@ protected void doExecute(Task task, OpenPointInTimeRequest request, ActionListen
                     new ActionListenerResponseHandler<SearchPhaseResult>(phaseListener, ShardOpenReaderResponse::new)
                 );
             },
-            ActionListener.map(listener, r -> new OpenPointInTimeResponse(r.pointInTimeId()))
+            ActionListener.map(listener, r -> {
+                assert r.pointInTimeId() != null : r;
+                return new OpenPointInTimeResponse(r.pointInTimeId());
+            })
         );
     }
 

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

@@ -10,7 +10,7 @@ dependencies {
 
 restResources {
   restApi {
-    includeXpack 'security', 'async_search', 'indices'
+    includeXpack 'security', 'async_search', 'indices', 'open_point_in_time', 'close_point_in_time'
   }
 }
 

x-pack/qa/multi-cluster-search-security/build.gradle

@@ -30,13 +30,15 @@
   - match: {indices.5.attributes.0: open}
   - match: {indices.6.name: my_remote_cluster:field_caps_index_3}
   - match: {indices.6.attributes.0: open}
-  - match: {indices.7.name: my_remote_cluster:secured_via_alias}
-  - match: {indices.7.attributes.0: open}
-  - match: {indices.8.name: my_remote_cluster:single_doc_index}
+  - match: {indices.7.name: my_remote_cluster:point_in_time_index }
+  - match: {indices.7.attributes.0: open }
+  - match: {indices.8.name: my_remote_cluster:secured_via_alias}
   - match: {indices.8.attributes.0: open}
-  - match: {indices.9.name: my_remote_cluster:test_index}
-  - match: {indices.9.aliases.0: aliased_test_index}
-  - match: {indices.9.attributes.0: open}
+  - match: {indices.9.name: my_remote_cluster:single_doc_index}
+  - match: {indices.10.attributes.0: open}
+  - match: {indices.10.name: my_remote_cluster:test_index}
+  - match: {indices.10.aliases.0: aliased_test_index}
+  - match: {indices.10.attributes.0: open}
   - match: {aliases.0.name: my_remote_cluster:.security}
   - match: {aliases.0.indices.0: .security-7}
   - match: {aliases.1.name: my_remote_cluster:aliased_closed_index}

x-pack/qa/multi-cluster-search-security/src/test/resources/rest-api-spec/test/multi_cluster/100_resolve_index.yml

@@ -0,0 +1,113 @@
+---
+setup:
+  - skip:
+      features: headers
+
+  - do:
+      cluster.health:
+        wait_for_status: yellow
+  - do:
+      security.put_user:
+        username: "joe"
+        body:  >
+          {
+            "password": "s3krit",
+            "roles" : [ "x_cluster_role" ]
+          }
+  - do:
+      security.put_role:
+        name: "x_cluster_role"
+        body:  >
+          {
+            "cluster": [],
+            "indices": [
+              {
+                "names": ["local_pit", "my_remote_cluster:point_in_time_index"],
+                "privileges": ["read"]
+              }
+            ]
+          }
+
+  - do:
+      security.put_user:
+        username: "remote"
+        body:  >
+          {
+            "password": "s3krit",
+            "roles" : [ "remote_ccs" ]
+          }
+  - do:
+      security.put_role:
+        name: "remote_ccs"
+        body:  >
+          {
+          }
+---
+teardown:
+  - do:
+      security.delete_user:
+        username: "joe"
+        ignore: 404
+  - do:
+      security.delete_role:
+        name: "x_cluster_role"
+        ignore: 404
+---
+"Search with point in time":
+
+  - do:
+      indices.create:
+        index: local_pit
+        body:
+          settings:
+            index:
+              number_of_shards: 2
+              number_of_replicas: 0
+          mappings:
+            properties:
+              created_at:
+                type: date
+                format: "yyyy-MM-dd"
+  - do:
+      bulk:
+        refresh: true
+        body:
+          - '{"index": {"_index": "local_pit"}}'
+          - '{"f": "l1", "created_at" : "2020-01-01"}'
+          - '{"index": {"_index": "local_pit"}}'
+          - '{"f": "l2", "created_at" : "2021-01-02"}'
+
+  - do:
+      headers: { Authorization: "Basic am9lOnMza3JpdA==" }
+      open_point_in_time:
+        index: my_remote_cluster:point_in_time_index,local_pit
+        keep_alive: 5m
+  - set: {id: pit_id}
+
+  - do:
+      headers: { Authorization: "Basic am9lOnMza3JpdA==" }
+      search:
+        rest_total_hits_as_int: true
+        sort: created_at
+        body:
+          query:
+            range:
+              created_at:
+                gte: "2020-01-03"
+          pit:
+            id: "$pit_id"
+            keep_alive: 1m
+
+  - match: { hits.total: 3 }
+  - match: { hits.hits.0._index: "my_remote_cluster:point_in_time_index" }
+  - match: { hits.hits.0._source.f: "r3" }
+  - match: { hits.hits.1._index: "my_remote_cluster:point_in_time_index" }
+  - match: { hits.hits.1._source.f: "r4" }
+  - match: { hits.hits.2._index: "local_pit" }
+  - match: { hits.hits.2._source.f: "l2" }
+
+  - do:
+      headers: { Authorization: "Basic am9lOnMza3JpdA==" }
+      close_point_in_time:
+        body:
+          id: "$pit_id"