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

Allow _shrink to N shards if source shards is a multiple of N #18699

Merged
merged 9 commits into from
Jun 7, 2016
Merged

Allow _shrink to N shards if source shards is a multiple of N #18699

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
10c0a4d
Allow `_shrink` to N shards if source shards is a multiple of N
s1monw Jun 1, 2016
e9d36b4
add javadocs and simplify code
s1monw Jun 3, 2016
69b346c
fix docs
s1monw Jun 3, 2016
63b2e00
qualify by class
s1monw Jun 3, 2016
4e92654
add additional tests for IndexMetaData
s1monw Jun 3, 2016
6cbad64
apply feedback
s1monw Jun 3, 2016
bb91175
fix expected shard size calculation for shard shrinking
s1monw Jun 6, 2016
c4061fb
make checkstyle happy
s1monw Jun 6, 2016
c7f7cdb
add extra ensureGreen()
s1monw Jun 7, 2016
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
2 changes: 1 addition & 1 deletion core/src/main/java/org/elasticsearch/action/admin/indices/shrink/ShrinkRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,7 +94,7 @@ public void setShrinkIndex(CreateIndexRequest shrinkIndexRequest) {
/**
* Returns the {@link CreateIndexRequest} for the shrink index
*/
public CreateIndexRequest getShrinkIndexReqeust() {
public CreateIndexRequest getShrinkIndexRequest() {
return shrinkIndexRequest;
}

Expand Down
2 changes: 1 addition & 1 deletion core/src/main/java/org/elasticsearch/action/admin/indices/shrink/ShrinkRequestBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ public ShrinkRequestBuilder setSourceIndex(String index) {
}

public ShrinkRequestBuilder setSettings(Settings settings) {
this.request.getShrinkIndexReqeust().settings(settings);
this.request.getShrinkIndexRequest().settings(settings);
return this;
}
}
56 changes: 31 additions & 25 deletions core/src/main/java/org/elasticsearch/action/admin/indices/shrink/TransportShrinkAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.admin.indices.create.CreateIndexClusterStateUpdateRequest;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.admin.indices.stats.IndexShardStats;
import org.elasticsearch.action.admin.indices.stats.IndicesStatsResponse;
import org.elasticsearch.action.support.ActionFilters;
import org.elasticsearch.action.support.master.TransportMasterNodeAction;
Expand All @@ -34,27 +35,17 @@
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
import org.elasticsearch.cluster.metadata.MetaDataCreateIndexService;
import org.elasticsearch.cluster.routing.IndexRoutingTable;
import org.elasticsearch.cluster.routing.ShardRouting;
import org.elasticsearch.cluster.routing.ShardRoutingState;
import org.elasticsearch.cluster.routing.allocation.AllocationService;
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.index.IndexNotFoundException;
import org.elasticsearch.index.shard.DocsStats;
import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.indices.IndexAlreadyExistsException;
import org.elasticsearch.threadpool.ThreadPool;
import org.elasticsearch.transport.TransportService;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Predicate;
import java.util.function.IntFunction;

/**
* Main class to initiate shrinking an index into a new index with a single shard
Expand Down Expand Up @@ -87,7 +78,7 @@ protected ShrinkResponse newResponse() {

@Override
protected ClusterBlockException checkBlock(ShrinkRequest request, ClusterState state) {
return state.blocks().indexBlockedException(ClusterBlockLevel.METADATA_WRITE, request.getShrinkIndexReqeust().index());
return state.blocks().indexBlockedException(ClusterBlockLevel.METADATA_WRITE, request.getShrinkIndexRequest().index());
}

@Override
Expand All @@ -98,7 +89,10 @@ protected void masterOperation(final ShrinkRequest shrinkRequest, final ClusterS
@Override
public void onResponse(IndicesStatsResponse indicesStatsResponse) {
CreateIndexClusterStateUpdateRequest updateRequest = prepareCreateIndexRequest(shrinkRequest, state,
indicesStatsResponse.getTotal().getDocs(), indexNameExpressionResolver);
(i) -> {
IndexShardStats shard = indicesStatsResponse.getIndex(sourceIndex).getIndexShards().get(i);
return shard == null ? null : shard.getPrimary().getDocs();
}, indexNameExpressionResolver);
createIndexService.createIndex(updateRequest, new ActionListener<ClusterStateUpdateResponse>() {
@Override
public void onResponse(ClusterStateUpdateResponse response) {
Expand Down Expand Up @@ -127,24 +121,36 @@ public void onFailure(Throwable e) {

// static for unittesting this method
static CreateIndexClusterStateUpdateRequest prepareCreateIndexRequest(final ShrinkRequest shrinkReqeust, final ClusterState state
, final DocsStats docsStats, IndexNameExpressionResolver indexNameExpressionResolver) {
, final IntFunction<DocsStats> perShardDocStats, IndexNameExpressionResolver indexNameExpressionResolver) {
final String sourceIndex = indexNameExpressionResolver.resolveDateMathExpression(shrinkReqeust.getSourceIndex());
final CreateIndexRequest targetIndex = shrinkReqeust.getShrinkIndexReqeust();
final CreateIndexRequest targetIndex = shrinkReqeust.getShrinkIndexRequest();
final String targetIndexName = indexNameExpressionResolver.resolveDateMathExpression(targetIndex.index());
final IndexMetaData metaData = state.metaData().index(sourceIndex);
final Settings targetIndexSettings = Settings.builder().put(targetIndex.settings())
.normalizePrefix(IndexMetaData.INDEX_SETTING_PREFIX).build();
long count = docsStats.getCount();
if (count >= IndexWriter.MAX_DOCS) {
throw new IllegalStateException("Can't merge index with more than [" + IndexWriter.MAX_DOCS
+ "] docs - too many documents");
int numShards = 1;
if (IndexMetaData.INDEX_NUMBER_OF_SHARDS_SETTING.exists(targetIndexSettings)) {
numShards = IndexMetaData.INDEX_NUMBER_OF_SHARDS_SETTING.get(targetIndexSettings);
}
for (int i = 0; i < numShards; i++) {
Set<ShardId> shardIds = IndexMetaData.selectShrinkShards(i, metaData, numShards);
long count = 0;
for (ShardId id : shardIds) {
DocsStats docsStats = perShardDocStats.apply(id.id());
if (docsStats != null) {
count += docsStats.getCount();
}
if (count > IndexWriter.MAX_DOCS) {
throw new IllegalStateException("Can't merge index with more than [" + IndexWriter.MAX_DOCS
+ "] docs - too many documents in shards " + shardIds);
}
}

}
targetIndex.cause("shrink_index");
targetIndex.settings(Settings.builder()
.put(targetIndexSettings)
// we can only shrink to 1 index so far!
.put("index.number_of_shards", 1)
);
Settings.Builder settingsBuilder = Settings.builder().put(targetIndexSettings);
settingsBuilder.put("index.number_of_shards", numShards);
targetIndex.settings(settingsBuilder);

return new CreateIndexClusterStateUpdateRequest(targetIndex,
"shrink_index", targetIndexName, true)
Expand Down
Loading