Skip to content

Commit a856aee

Browse files
jasontedorjasontedor
committed
Add dedicated test for chain replication
This commit adds a dedicated test that chain replication leader -> middle -> follow is successful.
1 parent 8e133ab commit a856aee

File tree

2 files changed

+239
-0
lines changed

2 files changed

+239
-0
lines changed

‎x-pack/plugin/ccr/qa/chain/build.gradle‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,61 @@
1+
import org.elasticsearch.gradle.test.RestIntegTestTask
2+
3+
apply plugin: 'elasticsearch.standalone-test'
4+
5+
dependencies {
6+
testCompile project(path: xpackModule('core'), configuration: 'testArtifacts')
7+
testCompile project(path: xpackModule('ccr'), configuration: 'runtime')
8+
}
9+
10+
task leaderClusterTest(type: RestIntegTestTask) {
11+
mustRunAfter(precommit)
12+
}
13+
14+
leaderClusterTestCluster {
15+
numNodes = 1
16+
clusterName = 'leader-cluster'
17+
setting 'xpack.license.self_generated.type', 'trial'
18+
setting 'node.name', 'leader'
19+
}
20+
21+
leaderClusterTestRunner {
22+
systemProperty 'tests.target_cluster', 'leader'
23+
}
24+
25+
task middleClusterTest(type: RestIntegTestTask) {}
26+
27+
middleClusterTestCluster {
28+
dependsOn leaderClusterTestRunner
29+
numNodes = 1
30+
clusterName = 'middle-cluster'
31+
setting 'xpack.license.self_generated.type', 'trial'
32+
setting 'cluster.remote.leader_cluster.seeds', "\"${-> leaderClusterTest.nodes.get(0).transportUri()}\""
33+
setting 'node.name', 'middle'
34+
}
35+
36+
middleClusterTestRunner {
37+
systemProperty 'tests.target_cluster', 'middle'
38+
systemProperty 'tests.leader_host', "${-> leaderClusterTest.nodes.get(0).httpUri()}"
39+
}
40+
41+
task followClusterTest(type: RestIntegTestTask) {}
42+
43+
followClusterTestCluster {
44+
dependsOn middleClusterTestRunner
45+
numNodes = 1
46+
clusterName = 'follow-cluster'
47+
setting 'xpack.license.self_generated.type', 'trial'
48+
setting 'cluster.remote.middle_cluster.seeds', "\"${-> middleClusterTest.nodes.get(0).transportUri()}\""
49+
setting 'node.name', 'follow'
50+
}
51+
52+
followClusterTestRunner {
53+
systemProperty 'tests.target_cluster', 'follow'
54+
systemProperty 'tests.leader_host', "${-> leaderClusterTest.nodes.get(0).httpUri()}"
55+
systemProperty 'tests.middle_host', "${-> middleClusterTest.nodes.get(0).httpUri()}"
56+
finalizedBy 'leaderClusterTestCluster#stop'
57+
finalizedBy 'middleClusterTestCluster#stop'
58+
}
59+
60+
check.dependsOn followClusterTest
61+
test.enabled = false

‎x-pack/plugin/ccr/qa/chain/src/test/java/org/elasticsearch/xpack/ccr/ChainIT.java‎

Lines changed: 178 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,178 @@
1+
/*
2+
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
3+
* or more contributor license agreements. Licensed under the Elastic License;
4+
* you may not use this file except in compliance with the Elastic License.
5+
*/
6+
package org.elasticsearch.xpack.ccr;
7+
8+
import org.apache.http.HttpHost;
9+
import org.apache.http.util.EntityUtils;
10+
import org.elasticsearch.client.Request;
11+
import org.elasticsearch.client.Response;
12+
import org.elasticsearch.client.RestClient;
13+
import org.elasticsearch.common.Strings;
14+
import org.elasticsearch.common.settings.Settings;
15+
import org.elasticsearch.common.xcontent.XContentBuilder;
16+
import org.elasticsearch.common.xcontent.XContentHelper;
17+
import org.elasticsearch.common.xcontent.json.JsonXContent;
18+
import org.elasticsearch.common.xcontent.support.XContentMapValues;
19+
import org.elasticsearch.test.rest.ESRestTestCase;
20+
21+
import java.io.IOException;
22+
import java.util.List;
23+
import java.util.Map;
24+
25+
import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
26+
import static org.hamcrest.Matchers.equalTo;
27+
28+
public class ChainIT extends ESRestTestCase {
29+
30+
private final String targetCluster = System.getProperty("tests.target_cluster");
31+
32+
@Override
33+
protected boolean preserveClusterUponCompletion() {
34+
return true;
35+
}
36+
37+
public void testFollowIndex() throws Exception {
38+
final int numDocs = 128;
39+
final String leaderIndexName = "leader";
40+
final String middleIndexName = "middle";
41+
if ("leader".equals(targetCluster)) {
42+
logger.info("Running against leader cluster");
43+
String mapping = "";
44+
if (randomBoolean()) { // randomly do source filtering on indexing
45+
mapping =
46+
"\"_doc\": {" +
47+
" \"_source\": {" +
48+
" \"includes\": [\"field\"]," +
49+
" \"excludes\": [\"filtered_field\"]" +
50+
" }"+
51+
"}";
52+
}
53+
Settings indexSettings = Settings.builder()
54+
.put("index.soft_deletes.enabled", true)
55+
.build();
56+
createIndex(leaderIndexName, indexSettings, mapping);
57+
for (int i = 0; i < numDocs; i++) {
58+
logger.info("Indexing doc [{}]", i);
59+
index(client(), leaderIndexName, Integer.toString(i), "field", i, "filtered_field", "true");
60+
}
61+
refresh(leaderIndexName);
62+
verifyDocuments(leaderIndexName, numDocs);
63+
} else if ("middle".equals(targetCluster)) {
64+
logger.info("Running against middle cluster");
65+
followIndex("leader_cluster:" + leaderIndexName, middleIndexName);
66+
assertBusy(() -> verifyDocuments(middleIndexName, numDocs));
67+
// unfollow and then follow and then index a few docs in leader index:
68+
pauseFollow(middleIndexName);
69+
resumeFollow("leader_cluster:" + leaderIndexName, middleIndexName);
70+
try (RestClient leaderClient = buildLeaderClient()) {
71+
int id = numDocs;
72+
index(leaderClient, leaderIndexName, Integer.toString(id), "field", id, "filtered_field", "true");
73+
index(leaderClient, leaderIndexName, Integer.toString(id + 1), "field", id + 1, "filtered_field", "true");
74+
index(leaderClient, leaderIndexName, Integer.toString(id + 2), "field", id + 2, "filtered_field", "true");
75+
}
76+
assertBusy(() -> verifyDocuments(middleIndexName, numDocs + 3));
77+
} else if ("follow".equals(targetCluster)) {
78+
logger.info("Running against follow cluster");
79+
final String followIndexName = "follow";
80+
followIndex("middle_cluster:" + middleIndexName, followIndexName);
81+
assertBusy(() -> verifyDocuments(followIndexName, numDocs + 3));
82+
83+
try (RestClient leaderClient = buildLeaderClient()) {
84+
int id = numDocs;
85+
index(leaderClient, leaderIndexName, Integer.toString(id), "field", id, "filtered_field", "true");
86+
index(leaderClient, leaderIndexName, Integer.toString(id + 1), "field", id + 1, "filtered_field", "true");
87+
index(leaderClient, leaderIndexName, Integer.toString(id + 2), "field", id + 2, "filtered_field", "true");
88+
}
89+
90+
try (RestClient middleClient = buildMiddleClient()) {
91+
assertBusy(() -> verifyDocuments(middleIndexName, numDocs + 6, middleClient));
92+
}
93+
94+
assertBusy(() -> verifyDocuments(followIndexName, numDocs + 6));
95+
} else {
96+
fail("unexpected target cluster [" + targetCluster + "]");
97+
}
98+
}
99+
100+
private static void index(RestClient client, String index, String id, Object... fields) throws IOException {
101+
XContentBuilder document = jsonBuilder().startObject();
102+
for (int i = 0; i < fields.length; i += 2) {
103+
document.field((String) fields[i], fields[i + 1]);
104+
}
105+
document.endObject();
106+
final Request request = new Request("POST", "/" + index + "/_doc/" + id);
107+
request.setJsonEntity(Strings.toString(document));
108+
assertOK(client.performRequest(request));
109+
}
110+
111+
private static void refresh(String index) throws IOException {
112+
assertOK(client().performRequest(new Request("POST", "/" + index + "/_refresh")));
113+
}
114+
115+
private static void resumeFollow(String leaderIndex, String followIndex) throws IOException {
116+
final Request request = new Request("POST", "/" + followIndex + "/_ccr/resume_follow");
117+
request.setJsonEntity("{\"leader_index\": \"" + leaderIndex + "\", \"poll_timeout\": \"10ms\"}");
118+
assertOK(client().performRequest(request));
119+
}
120+
121+
private static void followIndex(String leaderIndex, String followIndex) throws IOException {
122+
final Request request = new Request("PUT", "/" + followIndex + "/_ccr/follow");
123+
request.setJsonEntity("{\"leader_index\": \"" + leaderIndex + "\", \"poll_timeout\": \"10ms\"}");
124+
assertOK(client().performRequest(request));
125+
}
126+
127+
private static void pauseFollow(String followIndex) throws IOException {
128+
assertOK(client().performRequest(new Request("POST", "/" + followIndex + "/_ccr/pause_follow")));
129+
}
130+
131+
private static void verifyDocuments(String index, int expectedNumDocs) throws IOException {
132+
verifyDocuments(index, expectedNumDocs, client());
133+
}
134+
135+
private static void verifyDocuments(final String index, final int expectedNumDocs, final RestClient client) throws IOException {
136+
final Request request = new Request("GET", "/" + index + "/_search");
137+
request.addParameter("size", Integer.toString(expectedNumDocs));
138+
request.addParameter("sort", "field:asc");
139+
request.addParameter("q", "filtered_field:true");
140+
Map<String, ?> response = toMap(client.performRequest(request));
141+
142+
int numDocs = (int) XContentMapValues.extractValue("hits.total", response);
143+
assertThat(numDocs, equalTo(expectedNumDocs));
144+
145+
List<?> hits = (List<?>) XContentMapValues.extractValue("hits.hits", response);
146+
assertThat(hits.size(), equalTo(expectedNumDocs));
147+
for (int i = 0; i < expectedNumDocs; i++) {
148+
int value = (int) XContentMapValues.extractValue("_source.field", (Map<?, ?>) hits.get(i));
149+
assertThat(i, equalTo(value));
150+
}
151+
}
152+
153+
private static Map<String, Object> toMap(Response response) throws IOException {
154+
return toMap(EntityUtils.toString(response.getEntity()));
155+
}
156+
157+
private static Map<String, Object> toMap(String response) {
158+
return XContentHelper.convertToMap(JsonXContent.jsonXContent, response, false);
159+
}
160+
161+
private RestClient buildLeaderClient() throws IOException {
162+
assert "leader".equals(targetCluster) == false;
163+
return buildClient(System.getProperty("tests.leader_host"));
164+
}
165+
166+
private RestClient buildMiddleClient() throws IOException {
167+
assert "middel".equals(targetCluster) == false;
168+
return buildClient(System.getProperty("tests.middle_host"));
169+
}
170+
171+
private RestClient buildClient(final String url) throws IOException {
172+
int portSeparator = url.lastIndexOf(':');
173+
HttpHost httpHost = new HttpHost(url.substring(0, portSeparator),
174+
Integer.parseInt(url.substring(portSeparator + 1)), getProtocol());
175+
return buildClient(Settings.EMPTY, new HttpHost[]{httpHost});
176+
}
177+
178+
}

0 commit comments

Comments
 (0)