Add ability to split shards #26931

s1monw · 2017-10-09T15:59:27Z

This change adds a new _split API that allows to split indices into a new
index with a power of two more shards that the source index. This API works
alongside the _shrink API but doesn't require any shard relocation before
indices can be split.

The split operation is conceptually an inverse _shrink operation since we
initialize the index with a syntetic number of routing shards that are used
for the consistent hashing at index time. Compared to indices created with
earlier versions this might produce slightly different shard distributions but
has no impact on the per-index backwards compatibility. For now, the user is
required to prepare an index to be splittable by setting the
index.number_of_routing_shards at index creation time. The setting allows the
user to prepare the index to be splittable in factors of
index.number_of_routing_shards ie. if the index is created with
index.number_of_routing_shards: 16 and index.number_of_shards: 2 it can be
split into 4, 8, 16 shards. This is an intermediate step until we can make
this the default. This also allows us to safely backport this change to 6.x.

The _split operation is implemented internally as a DeleteByQuery on the
lucene level that is executed while the primary shards execute their initial
recovery. Subsequent merges that are triggered due to this operation will not be
executed immediately. All merges will be deferred unti the shards are started
and will then be throttled accordingly.

This change is intended for the 6.1 feature release but will not support pre-6.1
indices to be split unless these indices have been shrunk before. In that case
these indices can be split backwards into their original number of shards.

This change adds a new `_split` API that allows to split indices into a new index with a power of two more shards that the source index. This API works alongside the `_split` API but doesn't require any shard relocation before indices can be split. The split operation is conceptually an inverse `_shrink` operation since we initialize the index with a _syntetic_ number of routing shards that are used for the consistent hashing at index time. Compared to indices created with earlier versions this might produce slightly different shard distributions but has no impact on the per-index backwards compatibility. For now, the user is required to prepare an index to be splittable by setting the split factor at index creation time. Users can decide by what factor they want to split the index ie. if an index should be splittable by into a multiple of 2 setting `index.routing_shards_factor: 1024` allows to split an index 10 times doubling the number of shards each time. This is an intermediate step until we can make this the default. This also allows us to safely backport this change to 6.x. The `_split` operation is implemented internally as a DeleteByQuery on the lucene level that is executed while the primary shards execute their initial recovery. Subsequent merges that are triggered due to this operation will not be executed immediately. All merges will be deferred unti the shards are started and will then be throttled accordingly. This change is intended for the 6.1 feature release but will not support pre-6.1 indices to be split unless these indices have been shrunk before. In that case these indices can be split backwards into their original number of shards.

s1monw · 2017-10-09T17:17:24Z

@ywelsch can you please review the allocation decider / StoreRecovery work
@jpountz please take a look at the lucene level parts
@bleskes / @nik9000 can you do a general review

dadoonet

I wonder if at some point we should have only one endpoint _resize instead of _split and _shrink?

s1monw · 2017-10-10T08:00:01Z

I wonder if at some point we should have only one endpoint _resize instead of _split and _shrink?

what's wrong with two endpoints? I makes it very clear what you wanna do no? there no implicit / hidden parameter you need to specify.

dadoonet · 2017-10-10T08:06:12Z

what's wrong with two endpoints?

Nothing. Just that the internal classes named Resize* made me think about it. Was just a thought.

s1monw · 2017-10-11T08:00:27Z

I took a look if I can break this PR into smaller pieces since it's quite big. I think it's possible to do several parts in core independently and I will open a branch for the guts. I will leave this one open for reference

jasontedor · 2017-10-11T12:21:38Z

@s1monw It's not clear to me that splitting this PR will make it easier to review. I think that this PR looks big, but it has a conceptual nut that once you grok, the rest of the PR is ceremony/piping around that, and docs and tests. I sometimes (not always) find these larger changes easier to digest if I can see the entire picture at once (like in the initial CCS PR).

jpountz

I left some comments about the query impl.

jpountz · 2017-10-11T13:39:07Z

core/src/main/java/org/elasticsearch/index/shard/ShardSplittingQuery.java

+     * This DISI visits every live doc and selects all docs that don't belong into this
+     * shard based on their id and rounting value. This is only used in a routing partitioned index.
+     */
+    private final class RoutingPartitionedDocIdSetIterator extends DocIdSetIterator {


It is theoretically illegal to use live docs in a query since it prevents caching on the core key. However you could implement this more efficiently with a two-phase iterator, and the good news is that Lucene will make sure to check live docs before calling matches(). You could pass a DocIdSetIterator.all(maxDoc) as an approximation.

good stuff I will explore this.

jpountz · 2017-10-11T13:40:59Z

core/src/main/java/org/elasticsearch/index/shard/ShardSplittingQuery.java

+                    return leftToVisit == 0 ? Status.STOP : Status.NO;
+
+            }
+        }


Do you need to support indices created in 5.x? If yes you will need to read the _uid field too (which is a string field).

FWIW indices will have either _id or __uid but never both.

yeah no we dont' suppport 5.x I will add comments

jpountz · 2017-10-11T13:44:01Z

core/src/main/java/org/elasticsearch/index/shard/ShardSplittingQuery.java

+                    return shardId == targetShardId;
+                };
+                if (terms == null) { // this is the common case - no partitioning and no _routing values
+                    findSplitDocs(IdFieldMapper.NAME, includeInShard, leafReader, bitSet::set);


beware that 5.x indices will have a _uid field rather than _id

yeah so I prevent this feature from being used on 5.x indices for that reason

I added a check to the ctor to make sure we are only running this on 6.0 onwards indices

jpountz · 2017-10-11T13:45:59Z

core/src/main/java/org/elasticsearch/index/shard/ShardSplittingQuery.java

+
+    @Override
+    public boolean equals(Object o) {
+        return sameClassAs(o);


Let's make equals/hashcode correct or throw an UOE so that we don't have caching bugs? Even if queries passed to IndexWriter.deleteDocuments are not supposed to be cached I'd like to make sure there are no issues if it occurs.

s1monw · 2017-10-11T14:20:20Z

@jasontedor fair enough. Lets do a review round here and see if it's ok.

Today we only allow to decode byte arrays where the data has a 0 offset and the same length as the array. Allowing to decode stuff from a slice will make decoding IDs cheaper if the the ID is for instance coming from a term dictionary or BytesRef. Relates to elastic#26931

s1monw · 2017-10-12T14:11:06Z

@jpountz I pushed changes for your comments here 68d2fa6

ywelsch

I've left a few comments, questions and suggestions.

ywelsch · 2017-10-11T14:36:24Z

core/src/main/java/org/elasticsearch/action/admin/indices/shrink/TransportResizeAction.java

+        return new CreateIndexClusterStateUpdateRequest(targetIndex,
+            cause, targetIndex.index(), targetIndexName, true)
+            // mappings are updated on the node when merging in the shards, this prevents race-conditions since all mapping must be
+            // applied once we took the snapshot and if somebody fucks things up and switches the index read/write and adds docs we miss


language please

good catch yannik!

ywelsch · 2017-10-12T14:28:40Z

core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestShrinkIndexAction.java

        shrinkIndexRequest.timeout(request.paramAsTime("timeout", shrinkIndexRequest.timeout()));
        shrinkIndexRequest.masterNodeTimeout(request.paramAsTime("master_timeout", shrinkIndexRequest.masterNodeTimeout()));
        shrinkIndexRequest.setWaitForActiveShards(ActiveShardCount.parseString(request.param("wait_for_active_shards")));
-        return channel -> client.admin().indices().shrinkIndex(shrinkIndexRequest, new AcknowledgedRestListener<ShrinkResponse>(channel) {
+        return channel -> client.admin().indices().resizeIndex(shrinkIndexRequest, new AcknowledgedRestListener<ResizeResponse>(channel) {


How does BWC work here for shrink?
If you send a shrink request to a v7.0 node in a mixed-version cluster where the master is v6.x, the master won't be able to understand the resize action?

There is a TransportShrinkAction that is registered to deal with the inverse situation. I have to do the same (register a dummy resize there) for 6.x in oder to allow this but this is not part of this PR. I can also just stick with shrink but that can be decided when we approach the backport

I added bwc code to the TransportResizeAction here all should be sorted out now. /cc @bleskes

ywelsch · 2017-10-12T14:34:43Z

core/src/main/java/org/elasticsearch/cluster/metadata/IndexMetaData.java

+        }
+        int numSourceShards = sourceIndexMetadata.getNumberOfShards();
+        if (numSourceShards > numTargetShards) {
+            throw new IllegalArgumentException("the number of source shards must be less that the number of target shards");


can you add numSourceShards and numTargetShards to the message?

ywelsch · 2017-10-12T14:35:18Z

core/src/main/java/org/elasticsearch/cluster/metadata/IndexMetaData.java

+    }
+
+    /**
+     * Selects the source shards fro a local shard recovery. This might either be a split or a shrink operation.


ywelsch · 2017-10-12T14:36:38Z

core/src/main/java/org/elasticsearch/cluster/metadata/IndexMetaData.java

+        } else if (sourceIndexMetadata.getNumberOfShards() < numTargetShards) {
+            return Collections.singleton(selectSplitShard(shardId, sourceIndexMetadata, numTargetShards));
+        }
+        throw new IllegalArgumentException("can't select recover from shards if both indices have the same number of shards");


if we have shrink and split, why not zero-copy clone ;)

valid request, different PR IMO

ywelsch · 2017-10-12T15:47:33Z

.../main/java/org/elasticsearch/cluster/routing/allocation/decider/ResizeAllocationDecider.java

+                    return Decision.ALWAYS;
+                }
+                ShardId shardId = IndexMetaData.selectSplitShard(shardRouting.id(), sourceIndexMetaData, indexMetaData.getNumberOfShards());
+                ShardRouting sourceShardRouting = allocation.routingTable().shardRoutingTable(shardId).primaryShard();


use (the mutable) RoutingNodes instead here: allocation.routingNodes().activePrimary(shardRouting.shardId()).
The routingTable can be arbitrarily stale at this point and should practically never be used (I wanted to remove it from the RoutingAllocation interface, but there are a few usages that are difficult to replace by RoutingNodes).

ywelsch · 2017-10-12T15:48:17Z

.../main/java/org/elasticsearch/cluster/routing/allocation/decider/ResizeAllocationDecider.java

+                }
+                ShardId shardId = IndexMetaData.selectSplitShard(shardRouting.id(), sourceIndexMetaData, indexMetaData.getNumberOfShards());
+                ShardRouting sourceShardRouting = allocation.routingTable().shardRoutingTable(shardId).primaryShard();
+                if (sourceShardRouting.active() == false) {


not needed with the above change. just check for null

ywelsch · 2017-10-12T15:59:45Z

.../main/java/org/elasticsearch/cluster/routing/allocation/decider/ResizeAllocationDecider.java

+    public Decision canForceAllocatePrimary(ShardRouting shardRouting, RoutingNode node, RoutingAllocation allocation) {
+        assert shardRouting.primary() : "must not call canForceAllocatePrimary on a non-primary shard " + shardRouting;
+        // check if we have passed the maximum retry threshold through canAllocate,
+        // if so, we don't want to force the primary allocation here


this comment can be removed

ywelsch · 2017-10-12T16:27:24Z

core/src/main/java/org/elasticsearch/index/shard/IndexShard.java

                final List<IndexShard> startedShards = new ArrayList<>();
                final IndexService sourceIndexService = indicesService.indexService(mergeSourceIndex);
-                final int numShards = sourceIndexService != null ? sourceIndexService.getIndexSettings().getNumberOfShards() : -1;
+                final Set<ShardId> requiredShards = IndexMetaData.selectRecoverFromShards(shardId().id(),
+                    sourceIndexService.getMetaData(), indexMetaData.getNumberOfShards());


sourceIndexService could be null? (you're doing the null check in the next line)

ywelsch · 2017-10-12T16:37:48Z

.../main/java/org/elasticsearch/cluster/routing/allocation/decider/ResizeAllocationDecider.java

+                if (sourceShardRouting.active() == false) {
+                    return allocation.decision(Decision.NO, NAME, "source primary shard [%s] is not active", sourceShardRouting.shardId());
+                }
+                if (node != null) { // we might get called from the 2 param canAllocate method..


maybe we should also check if the ES version on the node is capable of splitting a shard.

good point I will add that

Today we only allow to decode byte arrays where the data has a 0 offset and the same length as the array. Allowing to decode stuff from a slice will make decoding IDs cheaper if the the ID is for instance coming from a term dictionary or BytesRef. Relates to #26931

This change adds a new `_split` API that allows to split indices into a new index with a power of two more shards that the source index. This API works alongside the `_shrink` API but doesn't require any shard relocation before indices can be split. The split operation is conceptually an inverse `_shrink` operation since we initialize the index with a _syntetic_ number of routing shards that are used for the consistent hashing at index time. Compared to indices created with earlier versions this might produce slightly different shard distributions but has no impact on the per-index backwards compatibility. For now, the user is required to prepare an index to be splittable by setting the `index.number_of_routing_shards` at index creation time. The setting allows the user to prepare the index to be splittable in factors of `index.number_of_routing_shards` ie. if the index is created with `index.number_of_routing_shards: 16` and `index.number_of_shards: 2` it can be split into `4, 8, 16` shards. This is an intermediate step until we can make this the default. This also allows us to safely backport this change to 6.x. The `_split` operation is implemented internally as a DeleteByQuery on the lucene level that is executed while the primary shards execute their initial recovery. Subsequent merges that are triggered due to this operation will not be executed immediately. All merges will be deferred unti the shards are started and will then be throttled accordingly. This change is intended for the 6.1 feature release but will not support pre-6.1 indices to be split unless these indices have been shrunk before. In that case these indices can be split backwards into their original number of shards.

Relates to #26931

* master: (25 commits) Disable bwc tests in preparation of backporting #26931 TemplateUpgradeService should only run on the master (#27294) Die with dignity while merging Fix profiling naming issues (#27133) Correctly encode warning headers Fixed references to Multi Index Syntax (#27283) Add an active Elasticsearch WordPress plugin link (#27279) Setting url parts as required to reflect the code base (#27263) keys in aggs percentiles need to be in quotes. (#26905) Align routing param type with search.json (#26958) Update to support bulk updates by query (#27172) Remove duplicated SnapshotStatus (#27276) add split index reference in indices.asciidoc Add ability to split shards (#26931) [Docs] Fix minor paragraph indentation error for multiple Indices params (#25535) Upgrade to Jackson 2.8.10 (#27230) Fix inconsistencies in the rest api specs for `tasks` (#27163) Adjust RestHighLevelClient method modifiers (#27238) Remove unused parameters in AnalysisRegistry (#27232) Add more information on `_failed_to_convert_` exception (#27034) ...

* ccr: (127 commits) Disable bwc tests in preparation of backporting elastic#26931 TemplateUpgradeService should only run on the master (elastic#27294) Die with dignity while merging Fix profiling naming issues (elastic#27133) Correctly encode warning headers Fixed references to Multi Index Syntax (elastic#27283) Add an active Elasticsearch WordPress plugin link (elastic#27279) Setting url parts as required to reflect the code base (elastic#27263) keys in aggs percentiles need to be in quotes. (elastic#26905) Align routing param type with search.json (elastic#26958) Update to support bulk updates by query (elastic#27172) Remove duplicated SnapshotStatus (elastic#27276) add split index reference in indices.asciidoc Add ability to split shards (elastic#26931) [Docs] Fix minor paragraph indentation error for multiple Indices params (elastic#25535) Upgrade to Jackson 2.8.10 (elastic#27230) Fix inconsistencies in the rest api specs for `tasks` (elastic#27163) Adjust RestHighLevelClient method modifiers (elastic#27238) Remove unused parameters in AnalysisRegistry (elastic#27232) Add more information on `_failed_to_convert_` exception (elastic#27034) ...

* 6.x: Update shrink's bwc version to 6.1.0 add split index reference in indices.asciidoc Add ability to split shards (#26931) TemplateUpgradeService should only run on the master (#27294) Die with dignity while merging Fix profiling naming issues (#27133) Correctly encode warning headers Fixed references to Multi Index Syntax (#27283) Add an active Elasticsearch WordPress plugin link (#27279) Setting url parts as required to reflect the code base (#27263) keys in aggs percentiles need to be in quotes. (#26905) Align routing param type with search.json (#26958) Update to support bulk updates by query (#27172) Remove duplicated SnapshotStatus (#27276)

bleskes · 2017-11-07T17:26:01Z

this is now backported to 6.x as well.

* master: (556 commits) Fix find remote when building BWC Remove colons from task and configuration names Add unreleased 5.6.5 version number testCreateSplitIndexToN: do not set `routing_partition_size` to >= `number_of_routing_shards` Snapshot/Restore: better handle incorrect chunk_size settings in FS repo (elastic#26844) Add limits for ngram and shingle settings (elastic#27211) (elastic#27318) Correct comment in index shard test Roll translog generation on primary promotion ObjectParser: Replace IllegalStateException with ParsingException (elastic#27302) scripted_metric _agg parameter disappears if params are provided (elastic#27159) Update discovery-ec2.asciidoc Update shrink's bwc version to 6.1.0 and enabled bwc tests Add limits for ngram and shingle settings (elastic#27211) Disable bwc tests in preparation of backporting elastic#26931 TemplateUpgradeService should only run on the master (elastic#27294) Die with dignity while merging Fix profiling naming issues (elastic#27133) Correctly encode warning headers Fixed references to Multi Index Syntax (elastic#27283) Add an active Elasticsearch WordPress plugin link (elastic#27279) ...

With elastic/elasticsearch#26931 the possibility for splitting shards was introduced. To make use of this feature for indices created with ES >=6.1 the config option `index.number_of_routing_shards` is required. This adds this config option currently set to 30 as it's a multiple of 1, 3 and 5, our current number of default shards in Beats and ES. This allows users with default configs to scale their split their shards. The `number_of_routing_shards` can also be overwritten in the config file.

While we have an assertion that checks if the number of routing shards is a multiple of the number of shards we need a real hard exception that checks this way earlier. This change adds a check and test that is executed before we create the index. Relates to elastic#26931

While we have an assertion that checks if the number of routing shards is a multiple of the number of shards we need a real hard exception that checks this way earlier. This change adds a check and test that is executed before we create the index. Relates to #26931

DiskThresholdDecider currently assumes that the source index of a resize operation (e.g. shrink) is available, and throws an IndexNotFoundException otherwise, thereby breaking any kind of shard allocation. This can be quite harmful if the source index is deleted during a shrink, or if the source index is unavailable during state recovery. While this behavior has been partly fixed in 6.1 and above (due to #26931), it relies on the order in which AllocationDeciders are executed (i.e. that ResizeAllocationDecider returns NO, ensuring that DiskThresholdDecider does not run, something that for example does not hold for the allocation explain API). This change adds a more complete fix, and also solves the situation for 5.6.

s1monw added :Core/Infra/Core Core issues without another label >enhancement release highlight v6.1.0 v7.0.0 labels Oct 9, 2017

s1monw added 2 commits October 9, 2017 19:09

Merge branch 'master' into split_shards

e166b1a

fix docs test

658d31a

s1monw requested review from jpountz, bleskes, ywelsch and nik9000 October 9, 2017 17:16

Merge branch 'master' into split_shards

be7ecd3

dadoonet reviewed Oct 10, 2017

View reviewed changes

jpountz reviewed Oct 11, 2017

View reviewed changes

Merge branch 'master' into split_shards

f13dfcd

s1monw mentioned this pull request Oct 12, 2017

Allow Uid#decodeId to decode from a byte array slice #26987

Merged

apply feedback from @jpoutz

68d2fa6

s1monw added 2 commits October 12, 2017 16:26

add a working hashcode / equals method to ShardSplittingQuery

e41efd8

fix test

7f33942

ywelsch reviewed Oct 12, 2017

View reviewed changes

bleskes added a commit that referenced this pull request Nov 7, 2017

Disable bwc tests in preparation of backporting #26931

2fc6c64

bleskes added a commit that referenced this pull request Nov 7, 2017

add split index reference in indices.asciidoc

9084243

Relates to #26931

ruflin mentioned this pull request Nov 14, 2017

Add number_of_routing_shards config set to 30 elastic/beats#5570

Merged

s1monw mentioned this pull request Nov 21, 2017

Protect shard splitting from illegal target shards #27468

Merged

javanna mentioned this pull request Nov 27, 2017

Implement shard splitting #8912

Closed

ywelsch mentioned this pull request Mar 30, 2018

Don't break allocation if resize source index is missing #29311

Merged

colings86 added v7.0.0-beta1 and removed v7.0.0 labels Feb 7, 2019

Add ability to split shards #26931

Add ability to split shards #26931

Conversation

s1monw commented Oct 9, 2017 • edited Loading

s1monw commented Oct 9, 2017

dadoonet left a comment

Choose a reason for hiding this comment

s1monw commented Oct 10, 2017

dadoonet commented Oct 10, 2017

s1monw commented Oct 11, 2017

jasontedor commented Oct 11, 2017

jpountz left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

s1monw commented Oct 11, 2017

s1monw commented Oct 12, 2017

ywelsch left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bleskes commented Nov 7, 2017

s1monw commented Oct 9, 2017 •

edited

Loading