ApsaraDB for MongoDB:How to check the progress of adding or removing shard nodes

For MongoDB versions earlier than 6.0

Before estimating task progress, obtain Balancer run statistics (success and failure counts) and chunk information for sharded collections awaiting migration.

You can obtain the Balancer's execution results and information about chunks of sharded tables pending migration using the following two methods:

• Method 1: Use output from the sh.status() command.

  Focus on two parts of the sh.status() output. The first part shows recent Balancer results, as in this example.

  ...
    balancer:
          Collections with active migrations: 
                  <db>.<collection> started at Wed Sep 27 2023 10:25:21 GMT+0800 (CST)
          Failed balancer rounds in last 5 attempts:  0
          Migration Results for the last 24 hours: 
                  300 : Success
    databases:
  ...

  The second part shows chunk details for sharded collections, as in this example.

  ...
  databases:
  ...
      {  "_id" : "<db>",  "primary" : "d-xxxxxxxxxxxxxxx3",  "partitioned" : true,  "version" : {  "uuid" : UUID("3409a337-c370-4425-ad72-8b8c6b0abd52"),  "lastMod" : 1 } }
            <db>.<collection>
                          shard key: { "<shard_key>" : "hashed" }
                          unique: false
                          balancing: true
                          chunks:
                                  d-xxxxxxxxxxxxxxx1      13630
                                  d-xxxxxxxxxxxxxxx2      13629
                                  d-xxxxxxxxxxxxxxx3      13652
                                  d-xxxxxxxxxxxxxxx4      13630
                                  d-xxxxxxxxxxxxxxx5      3719
                          too many chunks to print, use verbose if you want to force print
  ...

  In this example, “d-xxxxxxxxxxxxxxx5” is the newly added shard node. You can also run the getShardDistribution command in the database containing the sharded collection to obtain chunk distribution details, as shown below.

  use <db>
  db.<collection>.getShardDistribution()

• Method 2: Read relevant data directly from the config database.

  View chunk statistics aggregated by shard as follows.

  db.getSiblingDB("config").chunks.aggregate([{$group: {_id: "$shard", count: {$sum: 1}}}])

  To view chunks on a specific shard grouped by namespace, run the following command.

  db.getSiblingDB("config").chunks.aggregate([{$match: {shard: "d-xxxxxxxxxxxxxx"}},{$group: {_id: "$ns", count: {$sum: 1}}}])

  To count chunks successfully migrated to a specific shard node in the past 24 hours, run the following command.

  // details.to specifies the destination shard for chunk migration
  // Use ISODate to define the time range in the time field
  db.getSiblingDB("config").changelog.find({"what" : "moveChunk.commit", "details.to" : "d-xxxxxxxxxxxxx","time" : {"$gte": ISODate("2023-09-26T00:00:00")}}).count()

For MongoDB versions 6.0 and later

Before estimating task progress, focus on data distribution across shards. You can still use sh.status() output as a reference, but shift attention from uneven chunk counts to actual data volume distribution.

• Obtain sharded collection distribution and data balance information.

  Use the getShardDistribution command to obtain distribution details and focus on data balance. Example:

  mongos> db.xxx.getShardDistribution()
  Shard d-xxx at xxx
   data : 379.26MiB docs : 8277367 chunks : 1
   estimated data per chunk : 379.26MiB
   estimated docs per chunk : 8277367
  Shard d-xxx at xxx
   data : 379.11MiB docs : 8272852 chunks : 1
   estimated data per chunk : 379.11MiB
   estimated docs per chunk : 8272852
  Shard d-xxx at xxx
   data : 379.18MiB docs : 8275108 chunks : 1
   estimated data per chunk : 379.18MiB
   estimated docs per chunk : 8275108
  Totals
   data : 1.11GiB docs : 24825327 chunks : 3
   Shard d-xxx contains 33.34% data, 33.34% docs in cluster, avg obj size on shard : 48B
   Shard d-xxx contains 33.32% data, 33.32% docs in cluster, avg obj size on shard : 48B
   Shard d-xxx contains 33.33% data, 33.33% docs in cluster, avg obj size on shard : 48B

  You can also run the following command for more detailed distribution information (including document count, total size, orphaned document count, and orphaned size):

  db.getSiblingDB("admin").aggregate( [{ $shardedDataDistribution: { } },{ $match: { "ns": "<db>.<collection>" } }] ).pretty()

  The following example output shows noticeable data imbalance across shards.

  {
          "ns" : "<db>.<collection>",
          "shards" : [
                  {
                          "shardName" : "d-xxxxxxxxxxxxxxx1",
                          "numOrphanedDocs" : 0,
                          "numOwnedDocuments" : 504298920,
                          "ownedSizeBytes" : NumberLong("833101815840"),
                          "orphanedSizeBytes" : 0
                  },
                  {
                          "shardName" : "d-xxxxxxxxxxxxxxx2",
                          "numOrphanedDocs" : 0,
                          "numOwnedDocuments" : 250283901,
                          "ownedSizeBytes" : NumberLong("409714745937"),
                          "orphanedSizeBytes" : 0
                  },
                  {
                          "shardName" : "d-xxxxxxxxxxxxxxx3",
                          "numOrphanedDocs" : 0,
                          "numOwnedDocuments" : 109098088,
                          "ownedSizeBytes" : NumberLong("178157177704"),
                          "orphanedSizeBytes" : 0
                  },
                  {
                          "shardName" : "d-xxxxxxxxxxxxxxx4",
                          "numOrphanedDocs" : 0,
                          "numOwnedDocuments" : 382018055,
                          "ownedSizeBytes" : NumberLong("630329790750"),
                          "orphanedSizeBytes" : 0
                  }
            ]
  }

  From this output, you can determine that the instance has 4 shard nodes and the total data volume for the sharded collection <db>.<collection> is approximately 2051303530231 bytes (833101815840 + 409714745937 + 178157177704 + 630329790750 = 2051303530231), or about 1910.4 GB.

• View successfully migrated data volume over the past day.

  Run the following query:

  pipeline = [
      { '$match': {
          'what': 'moveChunk.commit',
      } },
      { '$group': {
          '_id': {
              'date': { '$dateToString': { 'format': '%Y-%m-%d', 'date': '$time'} },
           },
          'chunks_moved': { '$sum': 1 },
          'docs_moved': { '$sum': '$details.counts.cloned' },
          'bytes_moved': { '$sum': '$details.counts.clonedBytes' },
      } }, 
      { '$sort': { '_id.date': -1} },
  ]
  db.getSiblingDB("config").changelog.aggregate(pipeline)

  Example output:

  mongos> db.getSiblingDB("config").changelog.aggregate(pipeline)
  { "_id" : { "date" : "2024-09-24" }, "chunks_moved" : 91, "docs_moved" : NumberLong(33071040), "bytes_moved" : NumberLong("11532786734") }
  { "_id" : { "date" : "2024-09-23" }, "chunks_moved" : 294, "docs_moved" : NumberLong(109806635), "bytes_moved" : NumberLong("38294757573") }
  { "_id" : { "date" : "2024-09-22" }, "chunks_moved" : 737, "docs_moved" : NumberLong(266800041), "bytes_moved" : NumberLong("93047288531") }
  { "_id" : { "date" : "2024-09-21" }, "chunks_moved" : 1448, "docs_moved" : NumberLong(525308393), "bytes_moved" : NumberLong("183117965820") }
  { "_id" : { "date" : "2024-09-20" }, "chunks_moved" : 1459, "docs_moved" : NumberLong(530020503), "bytes_moved" : NumberLong("184390749022") }
  { "_id" : { "date" : "2024-09-19" }, "chunks_moved" : 1478, "docs_moved" : NumberLong(557833705), "bytes_moved" : NumberLong("186910196651") }
  { "_id" : { "date" : "2024-09-18" }, "chunks_moved" : 1360, "docs_moved" : NumberLong(508484825), "bytes_moved" : NumberLong("173486932689") }

  This output shows that on 2024-09-21, 183117965820 bytes (about 170.5 GB) were migrated.

For MongoDB versions earlier than 6.0

After obtaining the number of successfully migrated chunks and current data distribution, estimate overall task progress and expected completion time.

Assume the following ideal conditions during shard addition: total chunk count remains constant, shard count stays fixed (no concurrent shard additions or removals), service load remains stable, and Balancer parameters use default values. Using the example from Step 3 under For MongoDB versions earlier than 6.0, you know:

• There are 5 shard nodes.

• 300 chunks migrated successfully during the Balancer window.

• The sharded collection <db>.<collection> has 58260 total chunks (13630 + 13629 + 13652 + 13630 + 3719 = 58260).

Therefore, the following can be calculated:

• At equilibrium, each shard should hold 11652 chunks (58260 ÷ 5 = 11652).

• At the current migration rate, task completion requires another 26.4 days ((11652 – 3719) ÷ 300 ≈ 26.4).

• The shard addition task is 32% complete (3719 ÷ 11652 = 32%).

For MongoDB versions 6.0 and later

Assume the sharded cluster instance has only one collection requiring balancing. Using the example from Step 3 under For MongoDB versions 6.0 and later, you know:

• There are 4 shard nodes.

• On 2024-09-21, approximately 170.5 GB (183117965820 bytes) were migrated.

• The sharded collection <db>.<collection> has a total data volume of about 1910.4 GB (2051303530231 bytes).

Therefore, the following can be calculated:

• At equilibrium, each shard should hold roughly 477.6 GB (512825882557.75 bytes).

• The total data volume needing migration to reach equilibrium is 875559682949 bytes (815.4 GB), broken down as follows:

  Shard1 needs to migrate 320275933282.25 bytes (833101815840 – 512825882557.75).

  Shard2 needs to migrate 103111136620.75 bytes (409714745937 – 512825882557.75).

  Shard3 needs to migrate 334668704853.75 bytes (178157177704 – 512825882557.75).

  Shard4 needs to migrate 117503908192.25 bytes (630329790750 – 512825882557.75).

  (320275933282.25 + 103111136620.75 + 334668704853.75 + 117503908192.25 = 875559682949)

• At the current migration rate, equilibrium will be reached in about 4.8 days (815.4 GB ÷ 170.5 GB/day ≈ 4.8 days).