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 all 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
49 changes: 40 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,44 @@ 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) {
/*
* We need to arrange that the primary term on all the shards in the shrunken index is at least as large as
* the maximum primary term on all the shards in the source index. This ensures that we have correct
* document-level semantics regarding sequence numbers in the shrunken index.
*/
final IndexMetaData sourceMetaData = currentState.metaData().getIndexSafe(shrinkFromIndex);
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 +434,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
101 changes: 96 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 @@ -19,41 +19,57 @@

package org.elasticsearch.action.admin.indices.create;

import org.apache.lucene.index.CorruptIndexException;
import org.apache.lucene.search.Sort;
import org.apache.lucene.search.SortField;
import org.apache.lucene.search.SortedSetSelector;
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.index.IndexRequest;
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.IndexShardRoutingTable;
import org.elasticsearch.cluster.routing.Murmur3HashFunction;
import org.elasticsearch.cluster.routing.RoutingTable;
import org.elasticsearch.cluster.routing.ShardRouting;
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.Index;
import org.elasticsearch.index.IndexService;
import org.elasticsearch.index.query.TermsQueryBuilder;
import org.elasticsearch.index.shard.IndexShard;
import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.indices.IndicesService;
import org.elasticsearch.plugins.Plugin;
import org.elasticsearch.test.ESIntegTestCase;
import org.elasticsearch.test.InternalSettingsPlugin;
import org.elasticsearch.test.VersionUtils;

import java.util.Arrays;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
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.CoreMatchers.not;
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 +151,81 @@ 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(scaledRandomIntBetween(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();

// fail random primary shards to force primary terms to increase
final Index source = resolveIndex("source");
final int iterations = scaledRandomIntBetween(0, 16);
for (int i = 0; i < iterations; i++) {
final String node = randomSubsetOf(1, internalCluster().nodesInclude("source")).get(0);
final IndicesService indexServices = internalCluster().getInstance(IndicesService.class, node);
final IndexService indexShards = indexServices.indexServiceSafe(source);
for (final Integer shardId : indexShards.shardIds()) {
final IndexShard shard = indexShards.getShard(shardId);
if (shard.routingEntry().primary() && randomBoolean()) {
disableAllocation("source");
shard.failShard("test", new Exception("test"));
// this can not succeed until the shard is failed and a replica is promoted
int id = 0;
while (true) {
// find an ID that routes to the right shard, we will only index to the shard that saw a primary failure
final String s = Integer.toString(id);
final int hash = Math.floorMod(Murmur3HashFunction.hash(s), numberOfShards);
if (hash == shardId) {
final IndexRequest request =
new IndexRequest("source", "type", s).source("{ \"f\": \"" + s + "\"}", XContentType.JSON);
client().index(request).get();
break;
} else {
id++;
}
}
enableAllocation("source");
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