-
Notifications
You must be signed in to change notification settings - Fork 7
PCSM-203 Implement copy of chunk distribution from source #71
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.
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
base: main
Are you sure you want to change the base?
Changes from all commits
2fa0425
5b13dfc
a52ceda
4633503
a502680
e3f9886
faed272
960cb5d
44b73b1
672af59
9de90e2
1368c84
72fd82f
4bf3873
d378ab9
707f7cc
062ff96
c1f7828
497d9b1
5204646
3a6fbfa
7708ad8
c44afd4
2a6e4a2
9135310
68abdda
cf627bf
8cc7295
26b6132
fc0cfe3
ac4d48b
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
|
|
@@ -4,21 +4,23 @@ | |||||
|
|
||||||
| Sharding support is available starting with {{pcsm.full_name}} 0.7.0 and is currently in technical preview stage. We encourage you to try it out and share your feedback. This will help us improve the feature in future releases. | ||||||
|
|
||||||
| {{pcsm.full_name}} supports replication between sharded MongoDB clusters, enabling you to migrate or synchronize data from one sharded deployment to another. This capability allows you to migrate sharded clusters with minimal downtime and synchronize data between sharded clusters for testing or development purposes. | ||||||
| {{pcsm.full_name}} supports replication between sharded MongoDB clusters. You can use it to migrate data from one sharded deployment to another with minimal downtime, or to keep data synchronized for testing and development. | ||||||
|
|
||||||
| ## Overview | ||||||
|
|
||||||
| The workflow for sharded clusters is similar to replica sets. See [How {{pcsm.full_name}} works](intro.md#replication-workflows) for the complete workflow overview. The key difference is that {{pcsm.short}} connects to `mongos` instances on both the source and target clusters instead of replica set members. | ||||||
| The replication workflow for sharded clusters is similar to the workflow for replica sets. See [How {{pcsm.full_name}} works](intro.md#replication-workflows) for an overview of the replication stages. | ||||||
|
|
||||||
| Since {{pcsm.short}} connects through `mongos`, the cluster topology doesn't matter. This means the source and target clusters can have different numbers of shards. | ||||||
| For sharded deployments, {{pcsm.short}} connects to `mongos` on both the source and target clusters instead of connecting directly to individual shard members. The source and target can have different numbers of shards. | ||||||
|
|
||||||
| Also, {{pcsm.short}} replicates data and not metadata. This means chunk distribution as well as the primary shard name for a collection may differ on source and target clusters. | ||||||
| {{pcsm.short}} does not continuously replicate sharding metadata. For collections with a ranged shard key, it uses the source chunk boundaries to prepare the target before the initial clone begins. Changes to the chunk layout that occur later on the source are not replicated to the target. | ||||||
|
Check notice on line 15 in docs/sharding.md
|
||||||
|
|
||||||
| The primary shard assignment can also differ between the source and target clusters. | ||||||
|
|
||||||
| ## Prerequisites | ||||||
|
|
||||||
| * {{pcsm.full_name}} version 0.7.0 or later | ||||||
| * Source and target clusters must be sharded MongoDB deployments | ||||||
| * Both clusters must be running the same MongoDB version. Check [Version requirements](deployment.md#version-requirements) for more information about supported versions. | ||||||
| * Use {{pcsm.full_name}} 0.7.0 or later. Automatic source chunk boundary preparation requires version 0.10.0 or later. | ||||||
| * Both the source and target clusters must be sharded MongoDB deployments. | ||||||
|
Check notice on line 22 in docs/sharding.md
|
||||||
| * The source and target clusters must use a supported version combination. See [Cross-version replication](cross-version.md) for supported source and target versions. | ||||||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
Suggested change
|
||||||
|
|
||||||
| ## Connection string format | ||||||
|
|
||||||
|
|
@@ -36,17 +38,98 @@ | |||||
|
|
||||||
| ### Initial sync preparation | ||||||
|
|
||||||
| Before starting the initial sync, {{pcsm.short}} checks which collections are sharded on the source cluster and creates corresponding sharded collections on the destination cluster. The only sharding configuration preserved from the source cluster is the sharding key; all other sharding details are handled internally by the destination cluster. | ||||||
| For collections with a ranged shard key, {{pcsm.short}} uses the source chunk boundaries to pre-split the collection on the target before copying any documents. If the source and target have the same number of shards, PCSM preserves the source chunk ownership pattern. If the shard counts differ, PCSM uses the source boundaries and determines the chunk placement across the available target shards. | ||||||
|
Check notice on line 41 in docs/sharding.md
|
||||||
|
|
||||||
| Collections with a hashed shard key keep the initial chunk layout created by MongoDB when shardCollection runs on the target. | ||||||
|
|
||||||
| {{pcsm.short}} does not continuously replicate sharding metadata after the initial preparation. See [Chunk distribution](#chunk-distribution). | ||||||
|
Check notice on line 45 in docs/sharding.md
|
||||||
|
|
||||||
| ### Balancer operation | ||||||
|
|
||||||
| {{pcsm.full_name}} connects to source and target clusters via a `mongos` instance. Therefore, you do not need to disable the balancer on either the source or target cluster before starting replication. The target cluster's balancer continues to operate normally and manages chunk distribution according to its own sharding configuration and balancer settings. | ||||||
|
|
||||||
| ### Chunk distribution | ||||||
| For ranged shard keys, PCSM prepares the target using the source chunk boundaries before the clone begins. Chunk migrations, splits, and merges that occur later are not replicated between the clusters. Each cluster continues to manage its own chunk layout. See [Manage sharded cluster balancer :octicons-link-external-16:](https://www.mongodb.com/docs/manual/tutorial/manage-sharded-cluster-balancer/){:target="_blank"} in the MongoDB documentation. | ||||||
|
Check notice on line 51 in docs/sharding.md
|
||||||
|
|
||||||
| ### If the pre-split fails | ||||||
|
|
||||||
| If {{pcsm.short}} cannot prepare the chunk layout on the target, the initial sync fails. PCSM does not fall back to copying the data into an unsplit collection. | ||||||
|
Check notice on line 55 in docs/sharding.md
|
||||||
|
|
||||||
| Check the PCSM logs and resolve the reported problem on the target. Then start a new synchronization run from the initial sync stage. | ||||||
|
Check notice on line 57 in docs/sharding.md
|
||||||
|
|
||||||
| ## Chunk distribution | ||||||
|
|
||||||
| !!! admonition "Version added: 0.10.0" | ||||||
|
|
||||||
| During the initial sync, {{pcsm.short}} prepares the chunk distribution of a sharded collection before copying its documents. This happens automatically for every sharded collection, immediately after the collection is sharded on the target. There is no flag and nothing to configure. | ||||||
|
Check notice on line 63 in docs/sharding.md
|
||||||
|
|
||||||
| For an empty collection with a ranged shard key, MongoDB initially creates a single chunk that covers the full shard key range. If the clone starts with this layout, writes can be concentrated on one shard and the target balancer may need to redistribute the data later. See [Data partitioning with chunks :octicons-link-external-16:](https://www.mongodb.com/docs/manual/core/sharding-data-partitioning/){:target="_blank"} in the MongoDB documentation. | ||||||
|
Check notice on line 65 in docs/sharding.md
|
||||||
|
|
||||||
| To avoid this, {{pcsm.short}} recreates the source chunk boundaries on the target before copying the data. How those chunks are placed depends on whether the source and target have the same number of shards. | ||||||
|
Check notice on line 67 in docs/sharding.md
|
||||||
|
|
||||||
| Collections with a hashed shard key use the initial chunk layout created by MongoDB. See [Hashed shard keys](#hashed-shard-keys). | ||||||
|
|
||||||
| !!! note | ||||||
| {{pcsm.short}} uses the source chunk layout to prepare the target before the clone. It does not keep the chunk layouts on the two clusters synchronized. Chunk migrations, splits, or merges that happen later on the source are not reproduced on the target. The layouts can therefore change independently as each cluster's balancer runs. This is expected and does not indicate a replication problem. See [Balancer operation](#balancer-operation). | ||||||
|
Check notice on line 72 in docs/sharding.md
|
||||||
|
|
||||||
| ### Same number of shards | ||||||
|
|
||||||
| For a source collection with more than one chunk, if the source and target have the same number of shards, {{pcsm.short}} sorts the shard IDs in each cluster and pairs them by their position in the sorted lists. For example, the first source shard is paired with the first target shard, the second source shard with the second target shard, and so on. | ||||||
|
Check notice on line 76 in docs/sharding.md
|
||||||
|
|
||||||
| PCSM then recreates each source chunk boundary on the target and places the corresponding target chunk on the shard paired with the source shard that owns that chunk. | ||||||
|
Check notice on line 78 in docs/sharding.md
|
||||||
|
|
||||||
| ??? example "Same number of shards" | ||||||
|
|
||||||
| {{pcsm.short}} does not preserve chunk distribution information from the source cluster. The target cluster manages chunk distribution internally through its balancer. This means that after replication, chunks may be distributed differently on the target cluster compared to the source cluster, which is expected behavior. | ||||||
| ```{.text .no-copy} | ||||||
| Source shards: src-a, src-b | ||||||
| Target shards: tgt-a, tgt-b | ||||||
|
|
||||||
| Source layout: | ||||||
| [-∞, 100) -> src-a | ||||||
| [100, +∞) -> src-b | ||||||
|
|
||||||
| Target layout: | ||||||
| [-∞, 100) -> tgt-a | ||||||
| [100, +∞) -> tgt-b | ||||||
| ``` | ||||||
| In this example, `src-a` is paired with `tgt-a` and `src-b` with `tgt-b`. The target keeps the same chunk boundaries and ownership pattern as the source. | ||||||
|
|
||||||
| ### Different number of shards | ||||||
|
|
||||||
| For a source collection with more than one chunk, if the source and target have different numbers of shards, {{pcsm.short}} cannot map source chunk ownership directly to the target. | ||||||
|
Check notice on line 98 in docs/sharding.md
|
||||||
|
|
||||||
| Instead, {{pcsm.short}} estimates the size of each source chunk and processes the largest chunks first. It places each chunk on the target shard that currently has the smallest estimated amount of assigned data. | ||||||
|
Check notice on line 100 in docs/sharding.md
|
||||||
|
|
||||||
| {{pcsm.short}} keeps track of the estimated total for each target shard as it assigns chunks. It then recreates the source chunk boundaries on the target using the calculated placement. | ||||||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
Suggested change
|
||||||
|
|
||||||
| ??? example "Different number of shards" | ||||||
|
|
||||||
| ```{.text .no-copy} | ||||||
| Target shards: tgt-a, tgt-b | ||||||
| Source chunk sizes: 100 MB, 60 MB, 40 MB | ||||||
|
|
||||||
| 100 MB -> tgt-a | ||||||
| 60 MB -> tgt-b | ||||||
| 40 MB -> tgt-b | ||||||
|
|
||||||
| Final estimated placement: | ||||||
| tgt-a: 100 MB | ||||||
| tgt-b: 100 MB | ||||||
| ``` | ||||||
| Here, the 100 MB chunk is placed on `tgt-a` first. The 60 MB chunk goes to `tgt-b`, which has no data assigned yet. When the 40 MB chunk is processed, `tgt-b` still has less estimated data than `tgt-a`, so the chunk is also placed there. | ||||||
|
|
||||||
| ### Hashed shard keys | ||||||
|
|
||||||
| {{pcsm.short}} does not pre-split a collection whose shard key contains a hashed field. The target keeps the initial chunk layout MongoDB creates when `shardCollection` runs, and the target balancer manages it from there. | ||||||
|
Check notice on line 122 in docs/sharding.md
|
||||||
|
|
||||||
| ### Check the chunk distribution | ||||||
|
|
||||||
| To check how a replicated collection is distributed, connect to the target mongos and run: | ||||||
|
Check notice on line 126 in docs/sharding.md
|
||||||
|
|
||||||
| ```javascript | ||||||
| db.getSiblingDB('<database>').getCollection('<collection>').getShardDistribution() | ||||||
| ``` | ||||||
|
|
||||||
| Since the target cluster already has information about which collections are sharded, it handles sharding internally. {{pcsm.short}} does not interfere with the target cluster's sharding configuration or chunk distribution. | ||||||
| The command shows the data distribution across the target shards. | ||||||
|
|
||||||
| ## Usage | ||||||
|
|
||||||
|
|
||||||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This still requires both sides to be sharded, which is not correct. #67 rewrites these exact lines, so one of the two PRs has to give...
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.