Skip to content

Commit

Permalink
Support point in time cross cluster search (#61827)
Browse files Browse the repository at this point in the history
This commit integrates point in time into cross cluster search.

Relates #61062
Closes #61790
  • Loading branch information
dnhatn committed Sep 10, 2020
1 parent 808c868 commit aafb2cb
Show file tree
Hide file tree
Showing 11 changed files with 571 additions and 75 deletions.

Large diffs are not rendered by default.

Original file line number Diff line number Diff line change
Expand Up @@ -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()));
}
}

Expand Down Expand Up @@ -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) {
Expand All @@ -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();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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),
Expand Down Expand Up @@ -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")) {
Expand Down
Original file line number Diff line number Diff line change
@@ -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;
}
};
}
}
Original file line number Diff line number Diff line change
@@ -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();
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
Loading

0 comments on commit aafb2cb

Please sign in to comment.