Skip to content

Commit 85b6a98

Browse files
kimchykimchy
committed
A setting to auto expand the number of replicas of an index (based on data nodes), closes elastic#623.
1 parent f5a9f2d commit 85b6a98

File tree

2 files changed

+148
-14
lines changed

2 files changed

+148
-14
lines changed

‎modules/elasticsearch/src/main/java/org/elasticsearch/cluster/metadata/MetaDataUpdateSettingsService.java

Lines changed: 45 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -20,9 +20,7 @@
2020
package org.elasticsearch.cluster.metadata;
2121

2222
import org.elasticsearch.ElasticSearchIllegalArgumentException;
23-
import org.elasticsearch.cluster.ClusterService;
24-
import org.elasticsearch.cluster.ClusterState;
25-
import org.elasticsearch.cluster.ProcessedClusterStateUpdateTask;
23+
import org.elasticsearch.cluster.*;
2624
import org.elasticsearch.cluster.routing.RoutingTable;
2725
import org.elasticsearch.common.component.AbstractComponent;
2826
import org.elasticsearch.common.inject.Inject;
@@ -36,13 +34,56 @@
3634
/**
3735
* @author kimchy (shay.banon)
3836
*/
39-
public class MetaDataUpdateSettingsService extends AbstractComponent {
37+
public class MetaDataUpdateSettingsService extends AbstractComponent implements ClusterStateListener {
4038

4139
private final ClusterService clusterService;
4240

4341
@Inject public MetaDataUpdateSettingsService(Settings settings, ClusterService clusterService) {
4442
super(settings);
4543
this.clusterService = clusterService;
44+
this.clusterService.add(this);
45+
}
46+
47+
@Override public void clusterChanged(ClusterChangedEvent event) {
48+
// update an index with number of replicas based on data nodes if possible
49+
if (!event.state().nodes().localNodeMaster()) {
50+
return;
51+
}
52+
if (!event.nodesChanged()) {
53+
return;
54+
}
55+
for (final IndexMetaData indexMetaData : event.state().metaData()) {
56+
String autoExpandReplicas = indexMetaData.settings().get("index.auto_expand_replicas");
57+
if (autoExpandReplicas != null) {
58+
try {
59+
final int numberOfReplicas = event.state().nodes().dataNodes().size() - 1;
60+
61+
int min = Integer.parseInt(autoExpandReplicas.substring(0, autoExpandReplicas.indexOf('-')));
62+
int max;
63+
String sMax = autoExpandReplicas.substring(autoExpandReplicas.indexOf('-') + 1);
64+
if (sMax.equals("all")) {
65+
max = event.state().nodes().dataNodes().size() - 1;
66+
} else {
67+
max = Integer.parseInt(sMax);
68+
}
69+
70+
if (numberOfReplicas >= min && numberOfReplicas <= max) {
71+
Settings settings = ImmutableSettings.settingsBuilder().put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, numberOfReplicas).build();
72+
updateSettings(settings, new String[]{indexMetaData.index()}, new Listener() {
73+
@Override public void onSuccess() {
74+
logger.info("[{}] auto expanded replicas to [{}]", indexMetaData.index(), numberOfReplicas);
75+
}
76+
77+
@Override public void onFailure(Throwable t) {
78+
logger.warn("[{}] fail to auto expand replicas to [{}]", indexMetaData.index(), numberOfReplicas);
79+
}
80+
});
81+
}
82+
} catch (Exception e) {
83+
logger.warn("[{}] failed to parse auto expand replicas", e, indexMetaData.index());
84+
}
85+
}
86+
}
4687
}
4788

4889
public void updateSettings(final Settings pSettings, final String[] indices, final Listener listener) {

‎modules/test/integration/src/test/java/org/elasticsearch/test/integration/indices/settings/UpdateNumberOfReplicasTests.java

Lines changed: 103 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -49,16 +49,6 @@ public class UpdateNumberOfReplicasTests extends AbstractNodesTests {
4949
client1 = getClient1();
5050
client2 = getClient2();
5151

52-
createIndex();
53-
}
54-
55-
protected void createIndex() {
56-
logger.info("Creating index test");
57-
client1.admin().indices().create(createIndexRequest("test")).actionGet();
58-
}
59-
60-
protected String getConcreteIndexName() {
61-
return "test";
6252
}
6353

6454
@AfterMethod public void closeNodes() {
@@ -76,6 +66,9 @@ protected Client getClient2() {
7666
}
7767

7868
@Test public void simpleUpdateNumberOfReplicasTests() throws Exception {
69+
logger.info("Creating index test");
70+
client1.admin().indices().create(createIndexRequest("test")).actionGet();
71+
7972
logger.info("Running Cluster Health");
8073
ClusterHealthResponse clusterHealth = client1.admin().cluster().prepareHealth().setWaitForGreenStatus().execute().actionGet();
8174
logger.info("Done Cluster Health, status " + clusterHealth.status());
@@ -148,4 +141,104 @@ protected Client getClient2() {
148141
assertThat(countResponse.count(), equalTo(10l));
149142
}
150143
}
144+
145+
@Test public void testAutoExpandNumberOfReplicas0ToData() {
146+
logger.info("--> creating index test with auto expand replicas");
147+
client1.admin().indices().prepareCreate("test").setSettings(settingsBuilder().put("number_of_shards", 2).put("auto_expand_replicas", "0-all")).execute().actionGet();
148+
149+
logger.info("--> running cluster health");
150+
ClusterHealthResponse clusterHealth = client1.admin().cluster().prepareHealth().setWaitForGreenStatus().setWaitForActiveShards(4).execute().actionGet();
151+
logger.info("--> done cluster health, status " + clusterHealth.status());
152+
assertThat(clusterHealth.timedOut(), equalTo(false));
153+
assertThat(clusterHealth.status(), equalTo(ClusterHealthStatus.GREEN));
154+
assertThat(clusterHealth.indices().get("test").activePrimaryShards(), equalTo(2));
155+
assertThat(clusterHealth.indices().get("test").numberOfReplicas(), equalTo(1));
156+
assertThat(clusterHealth.indices().get("test").activeShards(), equalTo(4));
157+
158+
logger.info("--> add another node, should increase the number of replicas");
159+
startNode("node3");
160+
161+
logger.info("--> running cluster health");
162+
clusterHealth = client1.admin().cluster().prepareHealth().setWaitForGreenStatus().setWaitForActiveShards(6).execute().actionGet();
163+
logger.info("--> done cluster health, status " + clusterHealth.status());
164+
assertThat(clusterHealth.timedOut(), equalTo(false));
165+
assertThat(clusterHealth.status(), equalTo(ClusterHealthStatus.GREEN));
166+
assertThat(clusterHealth.indices().get("test").activePrimaryShards(), equalTo(2));
167+
assertThat(clusterHealth.indices().get("test").numberOfReplicas(), equalTo(2));
168+
assertThat(clusterHealth.indices().get("test").activeShards(), equalTo(6));
169+
170+
logger.info("--> closing one node");
171+
closeNode("node3");
172+
173+
logger.info("--> running cluster health");
174+
clusterHealth = client1.admin().cluster().prepareHealth().setWaitForGreenStatus().setWaitForActiveShards(4).execute().actionGet();
175+
logger.info("--> done cluster health, status " + clusterHealth.status());
176+
assertThat(clusterHealth.timedOut(), equalTo(false));
177+
assertThat(clusterHealth.status(), equalTo(ClusterHealthStatus.GREEN));
178+
assertThat(clusterHealth.indices().get("test").activePrimaryShards(), equalTo(2));
179+
assertThat(clusterHealth.indices().get("test").numberOfReplicas(), equalTo(1));
180+
assertThat(clusterHealth.indices().get("test").activeShards(), equalTo(4));
181+
182+
logger.info("--> closing another node");
183+
closeNode("node2");
184+
185+
logger.info("--> running cluster health");
186+
clusterHealth = client1.admin().cluster().prepareHealth().setWaitForGreenStatus().setWaitForActiveShards(2).execute().actionGet();
187+
logger.info("--> done cluster health, status " + clusterHealth.status());
188+
assertThat(clusterHealth.timedOut(), equalTo(false));
189+
assertThat(clusterHealth.status(), equalTo(ClusterHealthStatus.GREEN));
190+
assertThat(clusterHealth.indices().get("test").activePrimaryShards(), equalTo(2));
191+
assertThat(clusterHealth.indices().get("test").numberOfReplicas(), equalTo(0));
192+
assertThat(clusterHealth.indices().get("test").activeShards(), equalTo(2));
193+
}
194+
195+
@Test public void testAutoExpandNumberReplicas1ToData() {
196+
logger.info("--> creating index test with auto expand replicas");
197+
client1.admin().indices().prepareCreate("test").setSettings(settingsBuilder().put("number_of_shards", 2).put("auto_expand_replicas", "1-all")).execute().actionGet();
198+
199+
logger.info("--> running cluster health");
200+
ClusterHealthResponse clusterHealth = client1.admin().cluster().prepareHealth().setWaitForGreenStatus().setWaitForActiveShards(4).execute().actionGet();
201+
logger.info("--> done cluster health, status " + clusterHealth.status());
202+
assertThat(clusterHealth.timedOut(), equalTo(false));
203+
assertThat(clusterHealth.status(), equalTo(ClusterHealthStatus.GREEN));
204+
assertThat(clusterHealth.indices().get("test").activePrimaryShards(), equalTo(2));
205+
assertThat(clusterHealth.indices().get("test").numberOfReplicas(), equalTo(1));
206+
assertThat(clusterHealth.indices().get("test").activeShards(), equalTo(4));
207+
208+
logger.info("--> add another node, should increase the number of replicas");
209+
startNode("node3");
210+
211+
logger.info("--> running cluster health");
212+
clusterHealth = client1.admin().cluster().prepareHealth().setWaitForGreenStatus().setWaitForActiveShards(6).execute().actionGet();
213+
logger.info("--> done cluster health, status " + clusterHealth.status());
214+
assertThat(clusterHealth.timedOut(), equalTo(false));
215+
assertThat(clusterHealth.status(), equalTo(ClusterHealthStatus.GREEN));
216+
assertThat(clusterHealth.indices().get("test").activePrimaryShards(), equalTo(2));
217+
assertThat(clusterHealth.indices().get("test").numberOfReplicas(), equalTo(2));
218+
assertThat(clusterHealth.indices().get("test").activeShards(), equalTo(6));
219+
220+
logger.info("--> closing one node");
221+
closeNode("node3");
222+
223+
logger.info("--> running cluster health");
224+
clusterHealth = client1.admin().cluster().prepareHealth().setWaitForGreenStatus().setWaitForActiveShards(4).execute().actionGet();
225+
logger.info("--> done cluster health, status " + clusterHealth.status());
226+
assertThat(clusterHealth.timedOut(), equalTo(false));
227+
assertThat(clusterHealth.status(), equalTo(ClusterHealthStatus.GREEN));
228+
assertThat(clusterHealth.indices().get("test").activePrimaryShards(), equalTo(2));
229+
assertThat(clusterHealth.indices().get("test").numberOfReplicas(), equalTo(1));
230+
assertThat(clusterHealth.indices().get("test").activeShards(), equalTo(4));
231+
232+
logger.info("--> closing another node");
233+
closeNode("node2");
234+
235+
logger.info("--> running cluster health");
236+
clusterHealth = client1.admin().cluster().prepareHealth().setWaitForYellowStatus().setWaitForActiveShards(2).execute().actionGet();
237+
logger.info("--> done cluster health, status " + clusterHealth.status());
238+
assertThat(clusterHealth.timedOut(), equalTo(false));
239+
assertThat(clusterHealth.status(), equalTo(ClusterHealthStatus.YELLOW));
240+
assertThat(clusterHealth.indices().get("test").activePrimaryShards(), equalTo(2));
241+
assertThat(clusterHealth.indices().get("test").numberOfReplicas(), equalTo(1));
242+
assertThat(clusterHealth.indices().get("test").activeShards(), equalTo(2));
243+
}
151244
}

0 commit comments

Comments
 (0)