Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Initialize primary term for shrunk indices #25307

Merged
merged 4 commits into from
Jun 20, 2017
Merged

Initialize primary term for shrunk indices #25307

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
28eaa98
Initialize primary term for shrunk indices
jasontedor Jun 20, 2017
993943d
Formatting
jasontedor Jun 20, 2017
f4b69c2
Fail instead of restart
jasontedor Jun 20, 2017
73b3442
Comment
jasontedor Jun 20, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 35 additions & 9 deletions core/src/main/java/org/elasticsearch/cluster/metadata/MetaDataCreateIndexService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,7 @@
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiFunction;
import java.util.function.Predicate;
import java.util.stream.IntStream;

import static org.elasticsearch.action.support.ContextPreservingActionListener.wrapPreservingContext;
import static org.elasticsearch.cluster.metadata.IndexMetaData.SETTING_AUTO_EXPAND_REPLICAS;
Expand Down Expand Up @@ -340,19 +341,39 @@ public ClusterState execute(ClusterState currentState) throws Exception {
indexSettingsBuilder.put(IndexMetaData.SETTING_INDEX_PROVIDED_NAME, request.getProvidedName());
indexSettingsBuilder.put(SETTING_INDEX_UUID, UUIDs.randomBase64UUID());
final Index shrinkFromIndex = request.shrinkFrom();
int routingNumShards = IndexMetaData.INDEX_NUMBER_OF_SHARDS_SETTING.get(indexSettingsBuilder.build());;
if (shrinkFromIndex != null) {
prepareShrinkIndexSettings(currentState, mappings.keySet(), indexSettingsBuilder, shrinkFromIndex,
request.index());
IndexMetaData sourceMetaData = currentState.metaData().getIndexSafe(shrinkFromIndex);
final IndexMetaData.Builder tmpImdBuilder = IndexMetaData.builder(request.index());

final int routingNumShards;
if (shrinkFromIndex == null) {
routingNumShards = IndexMetaData.INDEX_NUMBER_OF_SHARDS_SETTING.get(indexSettingsBuilder.build());
} else {
final IndexMetaData sourceMetaData = currentState.metaData().getIndexSafe(shrinkFromIndex);
routingNumShards = sourceMetaData.getRoutingNumShards();
}
tmpImdBuilder.setRoutingNumShards(routingNumShards);

if (shrinkFromIndex != null) {
prepareShrinkIndexSettings(
currentState, mappings.keySet(), indexSettingsBuilder, shrinkFromIndex, request.index());
}
final Settings actualIndexSettings = indexSettingsBuilder.build();
tmpImdBuilder.settings(actualIndexSettings);

if (shrinkFromIndex != null) {
final IndexMetaData sourceMetaData = currentState.metaData().getIndexSafe(shrinkFromIndex);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can you please add a comment explaining why we do this?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I pushed a comment.

final long primaryTerm =
IntStream
.range(0, sourceMetaData.getNumberOfShards())
.mapToLong(sourceMetaData::primaryTerm)
.max()
.getAsLong();
for (int shardId = 0; shardId < tmpImdBuilder.numberOfShards(); shardId++) {
tmpImdBuilder.primaryTerm(shardId, primaryTerm);
}
}

Settings actualIndexSettings = indexSettingsBuilder.build();
IndexMetaData.Builder tmpImdBuilder = IndexMetaData.builder(request.index())
.setRoutingNumShards(routingNumShards);
// Set up everything, now locally create the index to see that things are ok, and apply
final IndexMetaData tmpImd = tmpImdBuilder.settings(actualIndexSettings).build();
final IndexMetaData tmpImd = tmpImdBuilder.build();
ActiveShardCount waitForActiveShards = request.waitForActiveShards();
if (waitForActiveShards == ActiveShardCount.DEFAULT) {
waitForActiveShards = tmpImd.getWaitForActiveShards();
Expand Down Expand Up @@ -408,6 +429,11 @@ public ClusterState execute(ClusterState currentState) throws Exception {
final IndexMetaData.Builder indexMetaDataBuilder = IndexMetaData.builder(request.index())
.settings(actualIndexSettings)
.setRoutingNumShards(routingNumShards);

for (int shardId = 0; shardId < tmpImd.getNumberOfShards(); shardId++) {
indexMetaDataBuilder.primaryTerm(shardId, tmpImd.primaryTerm(shardId));
}

for (MappingMetaData mappingMd : mappingsMetaData.values()) {
indexMetaDataBuilder.putMapping(mappingMd);
}
Expand Down
59 changes: 54 additions & 5 deletions core/src/test/java/org/elasticsearch/action/admin/indices/create/ShrinkIndexIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,23 +25,21 @@
import org.apache.lucene.search.SortedSetSortField;
import org.elasticsearch.Version;
import org.elasticsearch.action.admin.cluster.reroute.ClusterRerouteResponse;
import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
import org.elasticsearch.action.admin.indices.segments.IndexSegments;
import org.elasticsearch.action.admin.indices.segments.IndexShardSegments;
import org.elasticsearch.action.admin.indices.segments.IndicesSegmentResponse;
import org.elasticsearch.action.admin.indices.segments.ShardSegments;
import org.elasticsearch.action.admin.indices.settings.get.GetSettingsResponse;
import org.elasticsearch.action.support.ActiveShardCount;
import org.elasticsearch.client.Client;
import org.elasticsearch.cluster.ClusterInfoService;
import org.elasticsearch.cluster.InternalClusterInfoService;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.cluster.routing.RoutingTable;
import org.elasticsearch.cluster.routing.UnassignedInfo;
import org.elasticsearch.common.Priority;
import org.elasticsearch.common.collect.ImmutableOpenMap;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.engine.Segment;
import org.elasticsearch.index.query.TermsQueryBuilder;
import org.elasticsearch.plugins.Plugin;
import org.elasticsearch.test.ESIntegTestCase;
Expand All @@ -50,10 +48,14 @@

import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.stream.IntStream;

import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.greaterThanOrEqualTo;

public class ShrinkIndexIT extends ESIntegTestCase {

Expand Down Expand Up @@ -135,6 +137,53 @@ public void testCreateShrinkIndexToN() {
assertHitCount(client().prepareSearch("source").setSize(100).setQuery(new TermsQueryBuilder("foo", "bar")).get(), 20);
}

public void testShrinkIndexPrimaryTerm() throws Exception {
final List<Integer> factors = Arrays.asList(2, 3, 5, 7);
final List<Integer> numberOfShardsFactors = randomSubsetOf(randomIntBetween(1, factors.size()), factors);
final int numberOfShards = numberOfShardsFactors.stream().reduce(1, (x, y) -> x * y);
final int numberOfTargetShards = randomSubsetOf(numberOfShardsFactors).stream().reduce(1, (x, y) -> x * y);
internalCluster().ensureAtLeastNumDataNodes(2);
prepareCreate("source").setSettings(Settings.builder().put(indexSettings()).put("number_of_shards", numberOfShards)).get();

final ImmutableOpenMap<String, DiscoveryNode> dataNodes =
client().admin().cluster().prepareState().get().getState().nodes().getDataNodes();
assertThat(dataNodes.size(), greaterThanOrEqualTo(2));
final DiscoveryNode[] discoveryNodes = dataNodes.values().toArray(DiscoveryNode.class);
final String mergeNode = discoveryNodes[0].getName();
ensureGreen();

// restart random data nodes to force the primary term for some shards to increase
for (int i = 0; i < randomIntBetween(0, 16); i++) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why do we need up to 16 restarts? Maybe it's faster to fail shards by getting IndexShard instances from internalCluster? also this if loop calls randomIntBetween(0, 16) many times... I'm not sure that's what you intended

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sure, I pushed a change that does this.

internalCluster().restartRandomDataNode();
ensureGreen();
}
// relocate all shards to one node such that we can merge it.
final Settings.Builder prepareShrinkSettings =
Settings.builder().put("index.routing.allocation.require._name", mergeNode).put("index.blocks.write", true);
client().admin().indices().prepareUpdateSettings("source").setSettings(prepareShrinkSettings).get();
ensureGreen();

final IndexMetaData indexMetaData = indexMetaData(client(), "source");
final long beforeShrinkPrimaryTerm = IntStream.range(0, numberOfShards).mapToLong(indexMetaData::primaryTerm).max().getAsLong();

// now merge source into target
final Settings shrinkSettings =
Settings.builder().put("index.number_of_replicas", 0).put("index.number_of_shards", numberOfTargetShards).build();
assertAcked(client().admin().indices().prepareShrinkIndex("source", "target").setSettings(shrinkSettings).get());

ensureGreen();

final IndexMetaData afterShrinkIndexMetaData = indexMetaData(client(), "target");
for (int shardId = 0; shardId < numberOfTargetShards; shardId++) {
assertThat(afterShrinkIndexMetaData.primaryTerm(shardId), equalTo(beforeShrinkPrimaryTerm + 1));
}
}

private static IndexMetaData indexMetaData(final Client client, final String index) {
final ClusterStateResponse clusterStateResponse = client.admin().cluster().state(new ClusterStateRequest()).actionGet();
return clusterStateResponse.getState().metaData().index(index);
}

public void testCreateShrinkIndex() {
internalCluster().ensureAtLeastNumDataNodes(2);
Version version = VersionUtils.randomVersion(random());
Expand Down