diff options
Diffstat (limited to 'src/mongo/db/s/balancer/balancer_policy.cpp')
| -rw-r--r-- | src/mongo/db/s/balancer/balancer_policy.cpp | 404 |
1 files changed, 323 insertions, 81 deletions
diff --git a/src/mongo/db/s/balancer/balancer_policy.cpp b/src/mongo/db/s/balancer/balancer_policy.cpp index 96619b52c80..6d4c1be900a 100644 --- a/src/mongo/db/s/balancer/balancer_policy.cpp +++ b/src/mongo/db/s/balancer/balancer_policy.cpp @@ -37,6 +37,7 @@ #include "mongo/db/s/balancer/type_migration.h" #include "mongo/logv2/log.h" +#include "mongo/s/balancer_configuration.h" #include "mongo/s/catalog/type_shard.h" #include "mongo/s/catalog/type_tags.h" #include "mongo/s/grid.h" @@ -282,16 +283,19 @@ Status BalancerPolicy::isShardSuitableReceiver(const ClusterStatistics::ShardSta return Status::OK(); } -ShardId BalancerPolicy::_getLeastLoadedReceiverShard( +std::tuple<ShardId, int64_t> BalancerPolicy::_getLeastLoadedReceiverShard( const ShardStatisticsVector& shardStats, const DistributionStatus& distribution, + const boost::optional<CollectionDataSizeInfoForBalancing>& collDataSizeInfo, const string& tag, - const stdx::unordered_set<ShardId>& excludedShards) { + const stdx::unordered_set<ShardId>& availableShards) { ShardId best; - unsigned minChunks = numeric_limits<unsigned>::max(); + int64_t currentMin = numeric_limits<int64_t>::max(); + + const auto shouldBalanceAccordingToDataSize = collDataSizeInfo.has_value(); for (const auto& stat : shardStats) { - if (excludedShards.count(stat.shardId)) + if (!availableShards.count(stat.shardId)) continue; auto status = isShardSuitableReceiver(stat, tag); @@ -299,40 +303,68 @@ ShardId BalancerPolicy::_getLeastLoadedReceiverShard( continue; } - unsigned myChunks = distribution.numberOfChunksInShard(stat.shardId); - if (myChunks >= minChunks) { - continue; - } + if (shouldBalanceAccordingToDataSize) { + const auto& shardSizeIt = collDataSizeInfo->shardToDataSizeMap.find(stat.shardId); + if (shardSizeIt == collDataSizeInfo->shardToDataSizeMap.end()) { + // Skip if stats not available (may happen if add|remove shard during a round) + continue; + } - best = stat.shardId; - minChunks = myChunks; + int64_t shardSize = shardSizeIt->second; + if (shardSize < currentMin) { + best = stat.shardId; + currentMin = shardSize; + } + } else { + int64_t myChunks = distribution.numberOfChunksInShard(stat.shardId); + if (myChunks < currentMin) { + best = stat.shardId; + currentMin = myChunks; + } + } } - return best; + return {best, currentMin}; } -ShardId BalancerPolicy::_getMostOverloadedShard( +std::tuple<ShardId, int64_t> BalancerPolicy::_getMostOverloadedShard( const ShardStatisticsVector& shardStats, const DistributionStatus& distribution, + const boost::optional<CollectionDataSizeInfoForBalancing>& collDataSizeInfo, const string& chunkTag, - const stdx::unordered_set<ShardId>& excludedShards) { + const stdx::unordered_set<ShardId>& availableShards) { ShardId worst; - unsigned maxChunks = 0; + long long currentMax = numeric_limits<long long>::min(); + + const auto shouldBalanceAccordingToDataSize = collDataSizeInfo.has_value(); for (const auto& stat : shardStats) { - if (excludedShards.count(stat.shardId)) + if (!availableShards.count(stat.shardId)) continue; - const unsigned shardChunkCount = - distribution.numberOfChunksInShardWithTag(stat.shardId, chunkTag); - if (shardChunkCount <= maxChunks) - continue; + if (shouldBalanceAccordingToDataSize) { + const auto& shardSizeIt = collDataSizeInfo->shardToDataSizeMap.find(stat.shardId); + if (shardSizeIt == collDataSizeInfo->shardToDataSizeMap.end()) { + // Skip if stats not available (may happen if add|remove shard during a round) + continue; + } - worst = stat.shardId; - maxChunks = shardChunkCount; + const auto shardSize = shardSizeIt->second; + if (shardSize > currentMax) { + worst = stat.shardId; + currentMax = shardSize; + } + } else { + const unsigned shardChunkCount = + distribution.numberOfChunksInShardWithTag(stat.shardId, chunkTag); + if (shardChunkCount > currentMax) { + worst = stat.shardId; + currentMax = shardChunkCount; + } + } } - return worst; + return {worst, currentMax}; } // Returns a random integer in [0, max) using a uniform random distribution. @@ -400,10 +432,12 @@ MigrateInfo chooseRandomMigration(const ShardStatisticsVector& shardStats, MoveChunkRequest::ForceJumbo::kDoNotForce}; } -MigrateInfosWithReason BalancerPolicy::balance(const ShardStatisticsVector& shardStats, - const DistributionStatus& distribution, - stdx::unordered_set<ShardId>* usedShards, - bool forceJumbo) { +MigrateInfosWithReason BalancerPolicy::balance( + const ShardStatisticsVector& shardStats, + const DistributionStatus& distribution, + const boost::optional<CollectionDataSizeInfoForBalancing>& collDataSizeInfo, + stdx::unordered_set<ShardId>* availableShards, + bool forceJumbo) { vector<MigrateInfo> migrations; MigrationReason firstReason = MigrationReason::none; @@ -426,7 +460,7 @@ MigrateInfosWithReason BalancerPolicy::balance(const ShardStatisticsVector& shar if (!stat.isDraining) continue; - if (usedShards->count(stat.shardId)) + if (!availableShards->count(stat.shardId)) continue; const vector<ChunkType>& chunks = distribution.getChunks(stat.shardId); @@ -447,8 +481,8 @@ MigrateInfosWithReason BalancerPolicy::balance(const ShardStatisticsVector& shar const string tag = distribution.getTagForChunk(chunk); - const ShardId to = - _getLeastLoadedReceiverShard(shardStats, distribution, tag, *usedShards); + const auto [to, _] = _getLeastLoadedReceiverShard( + shardStats, distribution, collDataSizeInfo, tag, *availableShards); if (!to.isValid()) { if (migrations.empty()) { LOGV2_WARNING(21889, @@ -462,17 +496,44 @@ MigrateInfosWithReason BalancerPolicy::balance(const ShardStatisticsVector& shar } invariant(to != stat.shardId); - migrations.emplace_back( - to, distribution.nss(), chunk, MoveChunkRequest::ForceJumbo::kForceBalancer); + + auto maxChunkSizeBytes = [&]() -> boost::optional<int64_t> { + if (collDataSizeInfo.has_value()) { + return collDataSizeInfo->maxChunkSizeBytes; + } + return boost::none; + }(); + + if (collDataSizeInfo.has_value()) { + migrations.emplace_back( + to, + chunk.getShard(), + distribution.nss(), + chunk.getCollectionUUID(), + chunk.getMin(), + boost::none /* max */, + chunk.getVersion(), + // Always force jumbo chunks to be migrated off draining shards + MoveChunkRequest::ForceJumbo::kForceBalancer, + maxChunkSizeBytes); + } else { + migrations.emplace_back(to, + distribution.nss(), + chunk, + MoveChunkRequest::ForceJumbo::kForceBalancer, + maxChunkSizeBytes); + } + if (firstReason == MigrationReason::none) { firstReason = MigrationReason::drain; } - invariant(usedShards->insert(stat.shardId).second); - invariant(usedShards->insert(to).second); + invariant(availableShards->erase(stat.shardId)); + invariant(availableShards->erase(to)); break; } if (migrations.empty()) { + availableShards->erase(stat.shardId); LOGV2_WARNING(21890, "Unable to find any chunk to move from draining shard " "{shardId}. numJumboChunks: {numJumboChunks}", @@ -480,13 +541,17 @@ MigrateInfosWithReason BalancerPolicy::balance(const ShardStatisticsVector& shar "shardId"_attr = stat.shardId, "numJumboChunks"_attr = numJumboChunks); } + + if (availableShards->size() < 2) { + return std::make_pair(std::move(migrations), firstReason); + } } } // 2) Check for chunks, which are on the wrong shard and must be moved off of it if (!distribution.tags().empty()) { for (const auto& stat : shardStats) { - if (usedShards->count(stat.shardId)) + if (!availableShards->count(stat.shardId)) continue; const vector<ChunkType>& chunks = distribution.getChunks(stat.shardId); @@ -507,11 +572,12 @@ MigrateInfosWithReason BalancerPolicy::balance(const ShardStatisticsVector& shar "Chunk violates zone, but it is jumbo and cannot be moved", "chunk"_attr = redact(chunk.toString()), "zone"_attr = redact(tag)); + continue; } - const ShardId to = - _getLeastLoadedReceiverShard(shardStats, distribution, tag, *usedShards); + const auto [to, _] = _getLeastLoadedReceiverShard( + shardStats, distribution, collDataSizeInfo, tag, *availableShards); if (!to.isValid()) { if (migrations.empty()) { LOGV2_WARNING(21892, @@ -525,18 +591,47 @@ MigrateInfosWithReason BalancerPolicy::balance(const ShardStatisticsVector& shar } invariant(to != stat.shardId); - migrations.emplace_back(to, - distribution.nss(), - chunk, - forceJumbo ? MoveChunkRequest::ForceJumbo::kForceBalancer - : MoveChunkRequest::ForceJumbo::kDoNotForce); + + auto maxChunkSizeBytes = [&]() -> boost::optional<int64_t> { + if (collDataSizeInfo.has_value()) { + return collDataSizeInfo->maxChunkSizeBytes; + } + return boost::none; + }(); + + if (collDataSizeInfo.has_value()) { + migrations.emplace_back(to, + chunk.getShard(), + distribution.nss(), + chunk.getCollectionUUID(), + chunk.getMin(), + boost::none /* max */, + chunk.getVersion(), + forceJumbo + ? MoveChunkRequest::ForceJumbo::kForceBalancer + : MoveChunkRequest::ForceJumbo::kDoNotForce, + maxChunkSizeBytes); + } else { + migrations.emplace_back(to, + distribution.nss(), + chunk, + forceJumbo + ? MoveChunkRequest::ForceJumbo::kForceBalancer + : MoveChunkRequest::ForceJumbo::kDoNotForce, + maxChunkSizeBytes); + } + if (firstReason == MigrationReason::none) { firstReason = MigrationReason::zoneViolation; } - invariant(usedShards->insert(stat.shardId).second); - invariant(usedShards->insert(to).second); + invariant(availableShards->erase(stat.shardId)); + invariant(availableShards->erase(to)); break; } + + if (availableShards->size() < 2) { + return std::make_pair(std::move(migrations), firstReason); + } } } @@ -546,14 +641,22 @@ MigrateInfosWithReason BalancerPolicy::balance(const ShardStatisticsVector& shar tagsPlusEmpty.push_back(""); for (const auto& tag : tagsPlusEmpty) { - const size_t totalNumberOfChunksWithTag = - (tag.empty() ? distribution.totalChunks() : distribution.totalChunksWithTag(tag)); - size_t totalNumberOfShardsWithTag = 0; + int64_t totalDataSizeOfShardsWithZone = 0; for (const auto& stat : shardStats) { if (tag.empty() || stat.shardTags.count(tag)) { totalNumberOfShardsWithTag++; + if (collDataSizeInfo.has_value()) { + const auto& shardSizeIt = + collDataSizeInfo->shardToDataSizeMap.find(stat.shardId); + if (shardSizeIt == collDataSizeInfo->shardToDataSizeMap.end()) { + // Skip if stats not available (may happen if add|remove shard during a + // round) + continue; + } + totalDataSizeOfShardsWithZone += shardSizeIt->second; + } } } @@ -575,18 +678,47 @@ MigrateInfosWithReason BalancerPolicy::balance(const ShardStatisticsVector& shar continue; } - // Calculate the rounded optimal number of chunks per shard - const size_t idealNumberOfChunksPerShardForTag = - (size_t)std::roundf(totalNumberOfChunksWithTag / (float)totalNumberOfShardsWithTag); - - while (_singleZoneBalance(shardStats, - distribution, - tag, - idealNumberOfChunksPerShardForTag, - &migrations, - usedShards, - forceJumbo ? MoveChunkRequest::ForceJumbo::kForceBalancer - : MoveChunkRequest::ForceJumbo::kDoNotForce)) { + const int64_t idealDataSizePerShardForZone = + totalDataSizeOfShardsWithZone / totalNumberOfShardsWithTag; + + auto singleZoneBalance = [&]() { + if (collDataSizeInfo.has_value()) { + tassert(ErrorCodes::BadValue, + str::stream() + << "Total data size for shards in zone " << tag << " and collection " + << distribution.nss() << " must be greater or equal than zero but is " + << totalDataSizeOfShardsWithZone, + totalDataSizeOfShardsWithZone >= 0); + + if (totalDataSizeOfShardsWithZone == 0) { + // No data to balance within this zone + return false; + } + + return _singleZoneBalanceBasedOnDataSize( + shardStats, + distribution, + *collDataSizeInfo, + tag, + idealDataSizePerShardForZone, + &migrations, + availableShards, + forceJumbo ? MoveChunkRequest::ForceJumbo::kForceBalancer + : MoveChunkRequest::ForceJumbo::kDoNotForce); + } + + return _singleZoneBalanceBasedOnChunks( + shardStats, + distribution, + tag, + totalNumberOfShardsWithTag, + &migrations, + availableShards, + forceJumbo ? MoveChunkRequest::ForceJumbo::kForceBalancer + : MoveChunkRequest::ForceJumbo::kDoNotForce); + }; + + while (singleZoneBalance()) { if (firstReason == MigrationReason::none) { firstReason = MigrationReason::chunksImbalance; } @@ -602,8 +734,16 @@ boost::optional<MigrateInfo> BalancerPolicy::balanceSingleChunk( const DistributionStatus& distribution) { const string tag = distribution.getTagForChunk(chunk); - ShardId newShardId = - _getLeastLoadedReceiverShard(shardStats, distribution, tag, stdx::unordered_set<ShardId>()); + stdx::unordered_set<ShardId> availableShards; + std::transform(shardStats.begin(), + shardStats.end(), + std::inserter(availableShards, availableShards.end()), + [](const ClusterStatistics::ShardStatistics& shardStatistics) -> ShardId { + return shardStatistics.shardId; + }); + + const auto [newShardId, _] = _getLeastLoadedReceiverShard( + shardStats, distribution, boost::none /* collDataSizeInfo */, tag, availableShards); if (!newShardId.isValid() || newShardId == chunk.getShard()) { return boost::optional<MigrateInfo>(); } @@ -612,14 +752,21 @@ boost::optional<MigrateInfo> BalancerPolicy::balanceSingleChunk( newShardId, distribution.nss(), chunk, MoveChunkRequest::ForceJumbo::kDoNotForce); } -bool BalancerPolicy::_singleZoneBalance(const ShardStatisticsVector& shardStats, - const DistributionStatus& distribution, - const string& tag, - size_t idealNumberOfChunksPerShardForTag, - vector<MigrateInfo>* migrations, - stdx::unordered_set<ShardId>* usedShards, - MoveChunkRequest::ForceJumbo forceJumbo) { - const ShardId from = _getMostOverloadedShard(shardStats, distribution, tag, *usedShards); +bool BalancerPolicy::_singleZoneBalanceBasedOnChunks(const ShardStatisticsVector& shardStats, + const DistributionStatus& distribution, + const string& tag, + size_t totalNumberOfShardsWithTag, + vector<MigrateInfo>* migrations, + stdx::unordered_set<ShardId>* availableShards, + MoveChunkRequest::ForceJumbo forceJumbo) { + // Calculate the rounded optimal number of chunks per shard + const size_t totalNumberOfChunksWithTag = + (tag.empty() ? distribution.totalChunks() : distribution.totalChunksWithTag(tag)); + const size_t idealNumberOfChunksPerShardForTag = + (size_t)std::roundf(totalNumberOfChunksWithTag / (float)totalNumberOfShardsWithTag); + + const auto [from, fromSize] = + _getMostOverloadedShard(shardStats, distribution, boost::none, tag, *availableShards); if (!from.isValid()) return false; @@ -629,13 +776,11 @@ bool BalancerPolicy::_singleZoneBalance(const ShardStatisticsVector& shardStats, if (max <= idealNumberOfChunksPerShardForTag) return false; - const ShardId to = _getLeastLoadedReceiverShard(shardStats, distribution, tag, *usedShards); + const auto [to, toSize] = + _getLeastLoadedReceiverShard(shardStats, distribution, boost::none, tag, *availableShards); if (!to.isValid()) { if (migrations->empty()) { - LOGV2(21882, - "No available shards to take chunks for zone {zone}", - "No available shards to take chunks for zone", - "zone"_attr = tag); + LOGV2(21882, "No available shards to take chunks for zone", "zone"_attr = tag); } return false; } @@ -682,8 +827,8 @@ bool BalancerPolicy::_singleZoneBalance(const ShardStatisticsVector& shardStats, } migrations->emplace_back(to, distribution.nss(), chunk, forceJumbo); - invariant(usedShards->insert(chunk.getShard()).second); - invariant(usedShards->insert(to).second); + invariant(availableShards->erase(chunk.getShard())); + invariant(availableShards->erase(to)); return true; } @@ -702,6 +847,93 @@ bool BalancerPolicy::_singleZoneBalance(const ShardStatisticsVector& shardStats, return false; } +bool BalancerPolicy::_singleZoneBalanceBasedOnDataSize( + const ShardStatisticsVector& shardStats, + const DistributionStatus& distribution, + const CollectionDataSizeInfoForBalancing& collDataSizeInfo, + const string& tag, + const int64_t idealDataSizePerShardForZone, + vector<MigrateInfo>* migrations, + stdx::unordered_set<ShardId>* availableShards, + MoveChunkRequest::ForceJumbo forceJumbo) { + const auto [from, fromSize] = + _getMostOverloadedShard(shardStats, distribution, collDataSizeInfo, tag, *availableShards); + if (!from.isValid()) + return false; + + const auto [to, toSize] = _getLeastLoadedReceiverShard( + shardStats, distribution, collDataSizeInfo, tag, *availableShards); + if (!to.isValid()) { + if (migrations->empty()) { + LOGV2(6581600, "No available shards to take chunks for zone", "zone"_attr = tag); + } + return false; + } + + if (from == to) { + return false; + } + + LOGV2_DEBUG(7548100, + 1, + "Balancing single zone", + "namespace"_attr = distribution.nss().ns(), + "zone"_attr = tag, + "idealDataSizePerShardForZone"_attr = idealDataSizePerShardForZone, + "fromShardId"_attr = from, + "fromShardDataSize"_attr = fromSize, + "toShardId"_attr = to, + "toShardDataSize"_attr = toSize, + "maxChunkSizeBytes"_attr = collDataSizeInfo.maxChunkSizeBytes); + + if (fromSize <= idealDataSizePerShardForZone) { + return false; + } + + if (fromSize - toSize < 3 * collDataSizeInfo.maxChunkSizeBytes) { + // Do not balance if the collection's size differs too few between the chosen shards + return false; + } + + const vector<ChunkType>& chunks = distribution.getChunks(from); + + unsigned numJumboChunks = 0; + + for (const auto& chunk : chunks) { + if (distribution.getTagForChunk(chunk) != tag) + continue; + + if (chunk.getJumbo()) { + numJumboChunks++; + continue; + } + + migrations->emplace_back(to, + chunk.getShard(), + distribution.nss(), + chunk.getCollectionUUID(), + chunk.getMin(), + boost::none /* max */, + chunk.getVersion(), + forceJumbo, + collDataSizeInfo.maxChunkSizeBytes); + invariant(availableShards->erase(chunk.getShard())); + invariant(availableShards->erase(to)); + return true; + } + + if (numJumboChunks) { + LOGV2_WARNING(6581602, + "Shard has only jumbo chunks for this collection and cannot be balanced", + "namespace"_attr = distribution.nss().ns(), + "shardId"_attr = from, + "zone"_attr = tag, + "numJumboChunks"_attr = numJumboChunks); + } + + return false; +} + ZoneRange::ZoneRange(const BSONObj& a_min, const BSONObj& a_max, const std::string& _zone) : min(a_min.getOwned()), max(a_max.getOwned()), zone(_zone) {} @@ -712,7 +944,8 @@ string ZoneRange::toString() const { MigrateInfo::MigrateInfo(const ShardId& a_to, const NamespaceString& a_nss, const ChunkType& a_chunk, - const MoveChunkRequest::ForceJumbo a_forceJumbo) + const MoveChunkRequest::ForceJumbo a_forceJumbo, + boost::optional<int64_t> maxChunkSizeBytes) : nss(a_nss), uuid(a_chunk.getCollectionUUID()) { invariant(a_to.isValid()); @@ -723,6 +956,7 @@ MigrateInfo::MigrateInfo(const ShardId& a_to, maxKey = a_chunk.getMax(); version = a_chunk.getVersion(); forceJumbo = a_forceJumbo; + optMaxChunkSizeBytes = maxChunkSizeBytes; } MigrateInfo::MigrateInfo(const ShardId& a_to, @@ -730,15 +964,17 @@ MigrateInfo::MigrateInfo(const ShardId& a_to, const NamespaceString& a_nss, const UUID& a_uuid, const BSONObj& a_min, - const BSONObj& a_max, + const boost::optional<BSONObj>& a_max, const ChunkVersion& a_version, - const MoveChunkRequest::ForceJumbo a_forceJumbo) + const MoveChunkRequest::ForceJumbo a_forceJumbo, + boost::optional<int64_t> maxChunkSizeBytes) : nss(a_nss), uuid(a_uuid), minKey(a_min), maxKey(a_max), version(a_version), - forceJumbo(a_forceJumbo) { + forceJumbo(a_forceJumbo), + optMaxChunkSizeBytes(maxChunkSizeBytes) { invariant(a_to.isValid()); invariant(a_from.isValid()); @@ -772,6 +1008,10 @@ string MigrateInfo::toString() const { << ", to " << to; } +boost::optional<int64_t> MigrateInfo::getMaxChunkSizeBytes() const { + return optMaxChunkSizeBytes; +} + SplitInfo::SplitInfo(const ShardId& inShardId, const NamespaceString& inNss, const ChunkVersion& inCollectionVersion, @@ -855,13 +1095,15 @@ DataSizeInfo::DataSizeInfo(const ShardId& shardId, const ChunkRange& chunkRange, const ChunkVersion& version, const KeyPattern& keyPattern, - bool estimatedValue) + bool estimatedValue, + int64_t maxSize) : shardId(shardId), nss(nss), uuid(uuid), chunkRange(chunkRange), version(version), keyPattern(keyPattern), - estimatedValue(estimatedValue) {} + estimatedValue(estimatedValue), + maxSize(maxSize) {} } // namespace mongo |