summaryrefslogtreecommitdiff
path: root/src/mongo/db/s/balancer/balancer_policy.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'src/mongo/db/s/balancer/balancer_policy.cpp')
-rw-r--r--src/mongo/db/s/balancer/balancer_policy.cpp811
1 files changed, 221 insertions, 590 deletions
diff --git a/src/mongo/db/s/balancer/balancer_policy.cpp b/src/mongo/db/s/balancer/balancer_policy.cpp
index d798aa7612e..96619b52c80 100644
--- a/src/mongo/db/s/balancer/balancer_policy.cpp
+++ b/src/mongo/db/s/balancer/balancer_policy.cpp
@@ -37,7 +37,6 @@
#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"
@@ -61,143 +60,65 @@ namespace {
// optimal average across all shards for a zone for a rebalancing migration to be initiated.
const size_t kDefaultImbalanceThreshold = 1;
-ChunkType makeChunkType(const UUID& collUUID, const Chunk& chunk) {
- ChunkType ct{collUUID, chunk.getRange(), chunk.getLastmod(), chunk.getShardId()};
- ct.setJumbo(chunk.isJumbo());
- return ct;
-}
-
-/**
- * Return a vector of zones after they have been normalized according to the given chunk
- * configuration.
- *
- * If a zone covers only partially a chunk, boundaries of that zone will be shrank so that the
- * normalized zone won't overlap with that chunk. The boundaries of a normalized zone will never
- * fall in the middle of a chunk.
- *
- * Additionally the vector will contain also zones for the "NoZone",
- */
-std::vector<ZoneRange> normalizeZones(const ChunkManager& cm, const ZoneInfo& zoneInfo) {
- std::vector<ZoneRange> normalizedRanges;
-
- auto lastMax = cm.getShardKeyPattern().getKeyPattern().globalMin();
-
- for (const auto& [max, zoneRange] : zoneInfo.zoneRanges()) {
- const auto& minChunk = cm.findIntersectingChunkWithSimpleCollation(zoneRange.min);
- const auto gtMin =
- SimpleBSONObjComparator::kInstance.evaluate(zoneRange.min > minChunk.getMin());
- const auto& normalizedMin = gtMin ? minChunk.getMax() : zoneRange.min;
-
-
- const auto& maxChunk = cm.findIntersectingChunkWithSimpleCollation(zoneRange.max);
- const auto gtMax =
- SimpleBSONObjComparator::kInstance.evaluate(zoneRange.max > maxChunk.getMin()) &&
- SimpleBSONObjComparator::kInstance.evaluate(
- zoneRange.max != cm.getShardKeyPattern().getKeyPattern().globalMax());
- const auto& normalizedMax = gtMax ? maxChunk.getMin() : zoneRange.max;
-
-
- if (SimpleBSONObjComparator::kInstance.evaluate(normalizedMin == normalizedMax)) {
- // This normalised zone has a length of zero, therefore can't contain any chunks so we
- // can ignore it
- continue;
- }
-
- if (SimpleBSONObjComparator::kInstance.evaluate(normalizedMin != lastMax)) {
- // The zone is not contiguous with the previous one so we add a kNoZoneRange
- // does not fully contain any chunk so we will ignore it
- normalizedRanges.emplace_back(lastMax, normalizedMin, ZoneInfo::kNoZoneName);
- }
-
- normalizedRanges.emplace_back(normalizedMin, normalizedMax, zoneRange.zone);
- lastMax = normalizedMax;
- }
-
- const auto& globalMaxKey = cm.getShardKeyPattern().getKeyPattern().globalMax();
- if (SimpleBSONObjComparator::kInstance.evaluate(lastMax != globalMaxKey)) {
- normalizedRanges.emplace_back(lastMax, globalMaxKey, ZoneInfo::kNoZoneName);
- }
- return normalizedRanges;
-}
-
} // namespace
-DistributionStatus::DistributionStatus(NamespaceString nss,
- ZoneInfo zoneInfo,
- const ChunkManager& chunkMngr)
- : _nss(std::move(nss)), _zoneInfo(std::move(zoneInfo)), _chunkMngr(chunkMngr) {
+DistributionStatus::DistributionStatus(NamespaceString nss, ShardToChunksMap shardToChunksMap)
+ : _nss(std::move(nss)), _shardChunks(std::move(shardToChunksMap)) {}
- _normalizedZones = normalizeZones(_chunkMngr, _zoneInfo);
-
- for (size_t zoneRangeIdx = 0; zoneRangeIdx < _normalizedZones.size(); zoneRangeIdx++) {
- const auto& zoneRange = _normalizedZones[zoneRangeIdx];
- chunkMngr.forEachOverlappingChunk(
- zoneRange.min, zoneRange.max, false /* isMaxInclusive */, [&](const auto& chunkInfo) {
- auto [zoneIt, created] =
- _shardZoneInfoMap[chunkInfo.getShardId().toString()].try_emplace(
- zoneRange.zone, 1 /* numChunks */, zoneRangeIdx, chunkInfo.getMin());
+size_t DistributionStatus::totalChunks() const {
+ size_t total = 0;
- if (!created) {
- ++(zoneIt->second.numChunks);
- }
- return true;
- });
+ for (const auto& shardChunk : _shardChunks) {
+ total += shardChunk.second.size();
}
+
+ return total;
}
size_t DistributionStatus::totalChunksWithTag(const std::string& tag) const {
size_t total = 0;
- for (const auto& [_, shardZoneInfo] : _shardZoneInfoMap) {
- const auto& zoneIt = shardZoneInfo.find(tag);
- if (zoneIt != shardZoneInfo.end()) {
- total += zoneIt->second.numChunks;
- }
+
+ for (const auto& shardChunk : _shardChunks) {
+ total += numberOfChunksInShardWithTag(shardChunk.first, tag);
}
+
return total;
}
size_t DistributionStatus::numberOfChunksInShard(const ShardId& shardId) const {
- const auto shardZonesIt = _shardZoneInfoMap.find(shardId.toString());
- if (shardZonesIt == _shardZoneInfoMap.end()) {
- return 0;
- }
- size_t total = 0;
- for (const auto& [_, shardZoneInfo] : shardZonesIt->second) {
- total += shardZoneInfo.numChunks;
- }
- return total;
+ const auto& shardChunks = getChunks(shardId);
+ return shardChunks.size();
}
size_t DistributionStatus::numberOfChunksInShardWithTag(const ShardId& shardId,
const string& tag) const {
- const auto shardZonesIt = _shardZoneInfoMap.find(shardId.toString());
- if (shardZonesIt == _shardZoneInfoMap.end()) {
- return 0;
- }
- const auto& shardTags = shardZonesIt->second;
+ const auto& shardChunks = getChunks(shardId);
+
+ size_t total = 0;
- const auto& zoneIt = shardTags.find(tag);
- if (zoneIt == shardTags.end()) {
- return 0;
+ for (const auto& chunk : shardChunks) {
+ if (tag == getTagForChunk(chunk)) {
+ total++;
+ }
}
- return zoneIt->second.numChunks;
+
+ return total;
}
-string DistributionStatus::getTagForRange(const ChunkRange& range) const {
- return _zoneInfo.getZoneForChunk(range);
+const vector<ChunkType>& DistributionStatus::getChunks(const ShardId& shardId) const {
+ ShardToChunksMap::const_iterator i = _shardChunks.find(shardId);
+ invariant(i != _shardChunks.end());
+
+ return i->second;
}
-const StringMap<ShardZoneInfo>& DistributionStatus::getZoneInfoForShard(
- const ShardId& shardId) const {
- static const StringMap<ShardZoneInfo> emptyMap;
- const auto shardZonesIt = _shardZoneInfoMap.find(shardId.toString());
- if (shardZonesIt == _shardZoneInfoMap.end()) {
- return emptyMap;
- }
- return shardZonesIt->second;
+Status DistributionStatus::addRangeToZone(const ZoneRange& range) {
+ return _zoneInfo.addRangeToZone(range);
}
-const string ZoneInfo::kNoZoneName = "";
+string DistributionStatus::getTagForChunk(const ChunkType& chunk) const {
+ return _zoneInfo.getZoneForChunk(chunk.getRange());
+}
ZoneInfo::ZoneInfo()
: _zoneRanges(SimpleBSONObjComparator::kInstance.makeBSONObjIndexedMap<ZoneRange>()) {}
@@ -249,11 +170,11 @@ string ZoneInfo::getZoneForChunk(const ChunkRange& chunk) const {
// We should never have a partial overlap with a chunk range. If it happens, treat it as if this
// chunk doesn't belong to a tag
if (minIntersect != maxIntersect) {
- return ZoneInfo::kNoZoneName;
+ return "";
}
if (minIntersect == _zoneRanges.end()) {
- return ZoneInfo::kNoZoneName;
+ return "";
}
const ZoneRange& intersectRange = minIntersect->second;
@@ -264,16 +185,17 @@ string ZoneInfo::getZoneForChunk(const ChunkRange& chunk) const {
return intersectRange.zone;
}
- return ZoneInfo::kNoZoneName;
+ return "";
}
+
/**
* read all tags for collection via the catalog client and add to the zoneInfo
*/
-StatusWith<ZoneInfo> createCollectionZoneInfo(OperationContext* opCtx,
- const NamespaceString& nss,
- const KeyPattern& keyPattern) {
- ZoneInfo zoneInfo;
+Status ZoneInfo::addTagsFromCatalog(OperationContext* opCtx,
+ const NamespaceString& nss,
+ const KeyPattern& keyPattern,
+ ZoneInfo& chunkMgr) {
const auto swCollectionTags =
Grid::get(opCtx)->catalogClient()->getTagsForCollection(opCtx, nss);
if (!swCollectionTags.isOK()) {
@@ -284,14 +206,16 @@ StatusWith<ZoneInfo> createCollectionZoneInfo(OperationContext* opCtx,
for (const auto& tag : collectionTags) {
auto status =
- zoneInfo.addRangeToZone(ZoneRange(keyPattern.extendRangeBound(tag.getMinKey(), false),
+ chunkMgr.addRangeToZone(ZoneRange(keyPattern.extendRangeBound(tag.getMinKey(), false),
keyPattern.extendRangeBound(tag.getMaxKey(), false),
tag.getTag()));
+
if (!status.isOK()) {
return status;
}
}
- return {std::move(zoneInfo)};
+
+ return Status::OK();
}
void DistributionStatus::report(BSONObjBuilder* builder) const {
@@ -299,15 +223,15 @@ void DistributionStatus::report(BSONObjBuilder* builder) const {
// Report all shards
BSONArrayBuilder shardArr(builder->subarrayStart("shards"));
- for (const auto& [shardId, zoneInfoMap] : _shardZoneInfoMap) {
+ for (const auto& shardChunk : _shardChunks) {
BSONObjBuilder shardEntry(shardArr.subobjStart());
- shardEntry.append("name", shardId);
+ shardEntry.append("name", shardChunk.first.toString());
- BSONObjBuilder tagsObj(shardEntry.subobjStart("tags"));
- for (const auto& [tagName, shardZoneInfo] : zoneInfoMap) {
- tagsObj.appendNumber(tagName, static_cast<long long>(shardZoneInfo.numChunks));
+ BSONArrayBuilder chunkArr(shardEntry.subarrayStart("chunks"));
+ for (const auto& chunk : shardChunk.second) {
+ chunkArr.append(chunk.toConfigBSON());
}
- tagsObj.doneFast();
+ chunkArr.doneFast();
shardEntry.doneFast();
}
@@ -350,7 +274,7 @@ Status BalancerPolicy::isShardSuitableReceiver(const ClusterStatistics::ShardSta
str::stream() << stat.shardId << " is currently draining."};
}
- if (chunkTag != ZoneInfo::kNoZoneName && !stat.shardTags.count(chunkTag)) {
+ if (!chunkTag.empty() && !stat.shardTags.count(chunkTag)) {
return {ErrorCodes::IllegalOperation,
str::stream() << stat.shardId << " is not in the correct zone " << chunkTag};
}
@@ -358,19 +282,16 @@ Status BalancerPolicy::isShardSuitableReceiver(const ClusterStatistics::ShardSta
return Status::OK();
}
-std::tuple<ShardId, int64_t> BalancerPolicy::_getLeastLoadedReceiverShard(
+ShardId BalancerPolicy::_getLeastLoadedReceiverShard(
const ShardStatisticsVector& shardStats,
const DistributionStatus& distribution,
- const boost::optional<CollectionDataSizeInfoForBalancing>& collDataSizeInfo,
const string& tag,
- const stdx::unordered_set<ShardId>& availableShards) {
+ const stdx::unordered_set<ShardId>& excludedShards) {
ShardId best;
- int64_t currentMin = numeric_limits<int64_t>::max();
-
- const auto shouldBalanceAccordingToDataSize = collDataSizeInfo.has_value();
+ unsigned minChunks = numeric_limits<unsigned>::max();
for (const auto& stat : shardStats) {
- if (!availableShards.count(stat.shardId))
+ if (excludedShards.count(stat.shardId))
continue;
auto status = isShardSuitableReceiver(stat, tag);
@@ -378,68 +299,40 @@ std::tuple<ShardId, int64_t> BalancerPolicy::_getLeastLoadedReceiverShard(
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;
- }
-
- 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;
- }
+ unsigned myChunks = distribution.numberOfChunksInShard(stat.shardId);
+ if (myChunks >= minChunks) {
+ continue;
}
+
+ best = stat.shardId;
+ minChunks = myChunks;
}
- return {best, currentMin};
+ return best;
}
-std::tuple<ShardId, int64_t> BalancerPolicy::_getMostOverloadedShard(
+ShardId BalancerPolicy::_getMostOverloadedShard(
const ShardStatisticsVector& shardStats,
const DistributionStatus& distribution,
- const boost::optional<CollectionDataSizeInfoForBalancing>& collDataSizeInfo,
const string& chunkTag,
- const stdx::unordered_set<ShardId>& availableShards) {
+ const stdx::unordered_set<ShardId>& excludedShards) {
ShardId worst;
- long long currentMax = numeric_limits<long long>::min();
-
- const auto shouldBalanceAccordingToDataSize = collDataSizeInfo.has_value();
+ unsigned maxChunks = 0;
for (const auto& stat : shardStats) {
- if (!availableShards.count(stat.shardId))
+ if (excludedShards.count(stat.shardId))
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;
- }
+ const unsigned shardChunkCount =
+ distribution.numberOfChunksInShardWithTag(stat.shardId, chunkTag);
+ if (shardChunkCount <= maxChunks)
+ continue;
- 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;
- }
- }
+ worst = stat.shardId;
+ maxChunks = shardChunkCount;
}
- return {worst, currentMax};
+ return worst;
}
// Returns a random integer in [0, max) using a uniform random distribution.
@@ -499,34 +392,18 @@ MigrateInfo chooseRandomMigration(const ShardStatisticsVector& shardStats,
"fromShardId"_attr = sourceShardId,
"toShardId"_attr = destShardId);
- const auto& randomChunk = [&] {
- const auto numChunksOnSourceShard = distribution.numberOfChunksInShard(sourceShardId);
- const auto rndChunkIdx = getRandomIndex(numChunksOnSourceShard);
- ChunkType rndChunk;
-
- int idx{0};
- distribution.getChunkManager().forEachChunk([&](const auto& chunk) {
- if (chunk.getShardId() == sourceShardId && idx++ == rndChunkIdx) {
- rndChunk = makeChunkType(distribution.getChunkManager().getUUID(), chunk);
- return false;
- }
- return true;
- });
-
- invariant(rndChunk.getShard().isValid());
- return rndChunk;
- }();
+ const auto& chunks = distribution.getChunks(sourceShardId);
- return {
- destShardId, distribution.nss(), randomChunk, MoveChunkRequest::ForceJumbo::kDoNotForce};
+ return {destShardId,
+ distribution.nss(),
+ chunks[getRandomIndex(chunks.size())],
+ MoveChunkRequest::ForceJumbo::kDoNotForce};
}
-MigrateInfosWithReason BalancerPolicy::balance(
- const ShardStatisticsVector& shardStats,
- const DistributionStatus& distribution,
- const boost::optional<CollectionDataSizeInfoForBalancing>& collDataSizeInfo,
- stdx::unordered_set<ShardId>* availableShards,
- bool forceJumbo) {
+MigrateInfosWithReason BalancerPolicy::balance(const ShardStatisticsVector& shardStats,
+ const DistributionStatus& distribution,
+ stdx::unordered_set<ShardId>* usedShards,
+ bool forceJumbo) {
vector<MigrateInfo> migrations;
MigrationReason firstReason = MigrationReason::none;
@@ -549,84 +426,50 @@ MigrateInfosWithReason BalancerPolicy::balance(
if (!stat.isDraining)
continue;
- if (!availableShards->count(stat.shardId))
+ if (usedShards->count(stat.shardId))
continue;
- // Now we know we need to move chunks off this shard, but only if permitted by the
+ const vector<ChunkType>& chunks = distribution.getChunks(stat.shardId);
+
+ if (chunks.empty())
+ continue;
+
+ // Now we know we need to move to chunks off this shard, but only if permitted by the
// tags policy
unsigned numJumboChunks = 0;
- const auto& shardZones = distribution.getZoneInfoForShard(stat.shardId);
- for (const auto& shardZone : shardZones) {
- const auto& zoneName = shardZone.first;
-
- const auto chunkFoundForShard = !distribution.forEachChunkOnShardInZone(
- stat.shardId, zoneName, [&](const auto& chunk) {
- if (chunk.isJumbo()) {
- numJumboChunks++;
- return true; // continue
- }
-
- const auto [to, _] = _getLeastLoadedReceiverShard(
- shardStats, distribution, collDataSizeInfo, zoneName, *availableShards);
- if (!to.isValid()) {
- if (migrations.empty()) {
- LOGV2_DEBUG(
- 21889,
- 3,
- "Chunk {chunk} is on a draining shard, but no appropriate "
- "recipient found",
- "Chunk is on a draining shard, but no appropriate "
- "recipient found",
- "chunk"_attr =
- redact(makeChunkType(
- distribution.getChunkManager().getUUID(), chunk)
- .toString()));
- }
- return true; // continue
- }
- invariant(to != stat.shardId);
-
- 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.getShardId(),
- distribution.nss(),
- distribution.getChunkManager().getUUID(),
- chunk.getMin(),
- boost::none /* max */,
- chunk.getLastmod(),
- // Always force jumbo chunks to be migrated off draining shards
- MoveChunkRequest::ForceJumbo::kForceBalancer,
- maxChunkSizeBytes);
- } else {
- migrations.emplace_back(
- to,
- distribution.nss(),
- makeChunkType(distribution.getChunkManager().getUUID(), chunk),
- MoveChunkRequest::ForceJumbo::kForceBalancer,
- maxChunkSizeBytes);
- }
-
- if (firstReason == MigrationReason::none) {
- firstReason = MigrationReason::drain;
- }
-
- invariant(availableShards->erase(stat.shardId));
- invariant(availableShards->erase(to));
- return false; // break
- });
-
- if (chunkFoundForShard) {
- break;
+ // Since we have to move all chunks, lets just do in order
+ for (const auto& chunk : chunks) {
+ if (chunk.getJumbo()) {
+ numJumboChunks++;
+ continue;
}
+
+ const string tag = distribution.getTagForChunk(chunk);
+
+ const ShardId to =
+ _getLeastLoadedReceiverShard(shardStats, distribution, tag, *usedShards);
+ if (!to.isValid()) {
+ if (migrations.empty()) {
+ LOGV2_WARNING(21889,
+ "Chunk {chunk} is on a draining shard, but no appropriate "
+ "recipient found",
+ "Chunk is on a draining shard, but no appropriate "
+ "recipient found",
+ "chunk"_attr = redact(chunk.toString()));
+ }
+ continue;
+ }
+
+ invariant(to != stat.shardId);
+ migrations.emplace_back(
+ to, distribution.nss(), chunk, MoveChunkRequest::ForceJumbo::kForceBalancer);
+ if (firstReason == MigrationReason::none) {
+ firstReason = MigrationReason::drain;
+ }
+ invariant(usedShards->insert(stat.shardId).second);
+ invariant(usedShards->insert(to).second);
+ break;
}
if (migrations.empty()) {
@@ -637,110 +480,62 @@ MigrateInfosWithReason BalancerPolicy::balance(
"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 (!availableShards->count(stat.shardId))
+ if (usedShards->count(stat.shardId))
continue;
- const auto& shardZones = distribution.getZoneInfoForShard(stat.shardId);
- for (const auto& shardZone : shardZones) {
- const auto& zoneName = shardZone.first;
+ const vector<ChunkType>& chunks = distribution.getChunks(stat.shardId);
+
+ for (const auto& chunk : chunks) {
+ const string tag = distribution.getTagForChunk(chunk);
+
+ if (tag.empty())
+ continue;
- if (zoneName == ZoneInfo::kNoZoneName)
+ if (stat.shardTags.count(tag))
continue;
- if (stat.shardTags.count(zoneName))
+ if (chunk.getJumbo()) {
+ LOGV2_WARNING(
+ 21891,
+ "Chunk {chunk} violates zone {zone}, but it is jumbo and cannot be moved",
+ "Chunk violates zone, but it is jumbo and cannot be moved",
+ "chunk"_attr = redact(chunk.toString()),
+ "zone"_attr = redact(tag));
continue;
+ }
- const auto chunkFoundForShard = !distribution.forEachChunkOnShardInZone(
- stat.shardId, zoneName, [&](const auto& chunk) {
- if (chunk.isJumbo()) {
- LOGV2_WARNING(
- 21891,
- "Chunk {chunk} violates zone {zone}, but it is jumbo and "
- "cannot be "
- "moved",
- "Chunk violates zone, but it is jumbo and cannot be moved",
- "chunk"_attr = redact(
- makeChunkType(distribution.getChunkManager().getUUID(), chunk)
- .toString()),
- "zone"_attr = redact(zoneName));
- return true; // continue
- }
-
- const auto [to, _] = _getLeastLoadedReceiverShard(
- shardStats, distribution, collDataSizeInfo, zoneName, *availableShards);
- if (!to.isValid()) {
- if (migrations.empty()) {
- LOGV2_DEBUG(
- 21892,
- 3,
- "Chunk {chunk} violates zone {zone}, but no appropriate "
- "recipient found",
- "Chunk violates zone, but no appropriate recipient found",
- "chunk"_attr =
- redact(makeChunkType(
- distribution.getChunkManager().getUUID(), chunk)
- .toString()),
- "zone"_attr = redact(zoneName));
- }
- return true; // continue
- }
- invariant(to != stat.shardId);
-
- 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.getShardId(),
- distribution.nss(),
- distribution.getChunkManager().getUUID(),
- chunk.getMin(),
- boost::none /* max */,
- chunk.getLastmod(),
- forceJumbo ? MoveChunkRequest::ForceJumbo::kForceBalancer
- : MoveChunkRequest::ForceJumbo::kDoNotForce,
- maxChunkSizeBytes);
- } else {
- migrations.emplace_back(
- to,
- distribution.nss(),
- makeChunkType(distribution.getChunkManager().getUUID(), chunk),
- forceJumbo ? MoveChunkRequest::ForceJumbo::kForceBalancer
- : MoveChunkRequest::ForceJumbo::kDoNotForce,
- maxChunkSizeBytes);
- }
-
- if (firstReason == MigrationReason::none) {
- firstReason = MigrationReason::zoneViolation;
- }
-
- invariant(availableShards->erase(stat.shardId));
- invariant(availableShards->erase(to));
- return false; // break
- });
-
- if (chunkFoundForShard) {
- break;
+ const ShardId to =
+ _getLeastLoadedReceiverShard(shardStats, distribution, tag, *usedShards);
+ if (!to.isValid()) {
+ if (migrations.empty()) {
+ LOGV2_WARNING(21892,
+ "Chunk {chunk} violates zone {zone}, but no appropriate "
+ "recipient found",
+ "Chunk violates zone, but no appropriate recipient found",
+ "chunk"_attr = redact(chunk.toString()),
+ "zone"_attr = redact(tag));
+ }
+ continue;
}
- }
- if (availableShards->size() < 2) {
- return std::make_pair(std::move(migrations), firstReason);
+
+ invariant(to != stat.shardId);
+ migrations.emplace_back(to,
+ distribution.nss(),
+ chunk,
+ forceJumbo ? MoveChunkRequest::ForceJumbo::kForceBalancer
+ : MoveChunkRequest::ForceJumbo::kDoNotForce);
+ if (firstReason == MigrationReason::none) {
+ firstReason = MigrationReason::zoneViolation;
+ }
+ invariant(usedShards->insert(stat.shardId).second);
+ invariant(usedShards->insert(to).second);
+ break;
}
}
}
@@ -748,32 +543,24 @@ MigrateInfosWithReason BalancerPolicy::balance(
// 3) for each tag balance
vector<string> tagsPlusEmpty(distribution.tags().begin(), distribution.tags().end());
- tagsPlusEmpty.push_back(ZoneInfo::kNoZoneName);
+ 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 == ZoneInfo::kNoZoneName || stat.shardTags.count(tag)) {
+ 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;
- }
}
}
// Skip zones which have no shards assigned to them. This situation is not harmful, but
// should not be possible so warn the operator to correct it.
if (totalNumberOfShardsWithTag == 0) {
- if (tag != ZoneInfo::kNoZoneName) {
+ if (!tag.empty()) {
LOGV2_WARNING(
21893,
"Zone {zone} in collection {namespace} has no assigned shards and chunks "
@@ -788,47 +575,18 @@ MigrateInfosWithReason BalancerPolicy::balance(
continue;
}
- 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()) {
+ // 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)) {
if (firstReason == MigrationReason::none) {
firstReason = MigrationReason::chunksImbalance;
}
@@ -842,18 +600,10 @@ boost::optional<MigrateInfo> BalancerPolicy::balanceSingleChunk(
const ChunkType& chunk,
const ShardStatisticsVector& shardStats,
const DistributionStatus& distribution) {
- const string tag = distribution.getTagForRange(chunk.getRange());
-
- 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);
+ const string tag = distribution.getTagForChunk(chunk);
+
+ ShardId newShardId =
+ _getLeastLoadedReceiverShard(shardStats, distribution, tag, stdx::unordered_set<ShardId>());
if (!newShardId.isValid() || newShardId == chunk.getShard()) {
return boost::optional<MigrateInfo>();
}
@@ -862,25 +612,14 @@ boost::optional<MigrateInfo> BalancerPolicy::balanceSingleChunk(
newShardId, distribution.nss(), chunk, MoveChunkRequest::ForceJumbo::kDoNotForce);
}
-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) {
- const auto totalNumberOfChunksWithTag = [&] {
- if (tag == ZoneInfo::kNoZoneName) {
- return static_cast<size_t>(distribution.getChunkManager().numChunks());
- }
- return 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);
+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);
if (!from.isValid())
return false;
@@ -890,11 +629,13 @@ bool BalancerPolicy::_singleZoneBalanceBasedOnChunks(const ShardStatisticsVector
if (max <= idealNumberOfChunksPerShardForTag)
return false;
- const auto [to, toSize] =
- _getLeastLoadedReceiverShard(shardStats, distribution, boost::none, tag, *availableShards);
+ const ShardId to = _getLeastLoadedReceiverShard(shardStats, distribution, tag, *usedShards);
if (!to.isValid()) {
if (migrations->empty()) {
- LOGV2(21882, "No available shards to take chunks for zone", "zone"_attr = tag);
+ LOGV2(21882,
+ "No available shards to take chunks for zone {zone}",
+ "No available shards to take chunks for zone",
+ "zone"_attr = tag);
}
return false;
}
@@ -927,34 +668,26 @@ bool BalancerPolicy::_singleZoneBalanceBasedOnChunks(const ShardStatisticsVector
if (imbalance < kDefaultImbalanceThreshold)
return false;
-
- const auto& fromShardId = from;
- const auto& toShardId = to;
+ const vector<ChunkType>& chunks = distribution.getChunks(from);
unsigned numJumboChunks = 0;
- const auto chunkFound =
- !distribution.forEachChunkOnShardInZone(fromShardId, tag, [&](const auto& chunk) {
- if (chunk.isJumbo()) {
- numJumboChunks++;
- return true; // continue
- }
+ for (const auto& chunk : chunks) {
+ if (distribution.getTagForChunk(chunk) != tag)
+ continue;
- migrations->emplace_back(toShardId,
- distribution.nss(),
- makeChunkType(distribution.getChunkManager().getUUID(), chunk),
- forceJumbo);
- invariant(availableShards->erase(chunk.getShardId()));
- invariant(availableShards->erase(toShardId));
- return false; // break
- });
-
- tassert(8236500,
- "Expected to find at least one chunk for shard '{}' in zone '{}'"_format(
- fromShardId.toString(), tag),
- chunkFound || numJumboChunks);
-
- if (!chunkFound && numJumboChunks) {
+ if (chunk.getJumbo()) {
+ numJumboChunks++;
+ continue;
+ }
+
+ migrations->emplace_back(to, distribution.nss(), chunk, forceJumbo);
+ invariant(usedShards->insert(chunk.getShard()).second);
+ invariant(usedShards->insert(to).second);
+ return true;
+ }
+
+ if (numJumboChunks) {
LOGV2_WARNING(
21894,
"Shard: {shardId}, collection: {namespace} has only jumbo chunks for "
@@ -966,99 +699,7 @@ bool BalancerPolicy::_singleZoneBalanceBasedOnChunks(const ShardStatisticsVector
"numJumboChunks"_attr = numJumboChunks);
}
- return chunkFound;
-}
-
-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 (toSize >= idealDataSizePerShardForZone) {
- // Do not use a shard if it already has more data than the ideal per-shard size
- 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 auto& fromShardId = from;
- const auto& toShardId = to;
-
- unsigned numJumboChunks = 0;
-
- const auto chunkFound =
- !distribution.forEachChunkOnShardInZone(fromShardId, tag, [&](const auto& chunk) {
- if (chunk.isJumbo()) {
- numJumboChunks++;
- return true; // continue
- }
-
- migrations->emplace_back(toShardId,
- chunk.getShardId(),
- distribution.nss(),
- distribution.getChunkManager().getUUID(),
- chunk.getMin(),
- boost::none /* max */,
- chunk.getLastmod(),
- forceJumbo,
- collDataSizeInfo.maxChunkSizeBytes);
- invariant(availableShards->erase(chunk.getShardId()));
- invariant(availableShards->erase(toShardId));
- return false; // break
- });
-
- if (!chunkFound && 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 chunkFound;
+ return false;
}
ZoneRange::ZoneRange(const BSONObj& a_min, const BSONObj& a_max, const std::string& _zone)
@@ -1071,8 +712,7 @@ string ZoneRange::toString() const {
MigrateInfo::MigrateInfo(const ShardId& a_to,
const NamespaceString& a_nss,
const ChunkType& a_chunk,
- const MoveChunkRequest::ForceJumbo a_forceJumbo,
- boost::optional<int64_t> maxChunkSizeBytes)
+ const MoveChunkRequest::ForceJumbo a_forceJumbo)
: nss(a_nss), uuid(a_chunk.getCollectionUUID()) {
invariant(a_to.isValid());
@@ -1083,7 +723,6 @@ 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,
@@ -1091,17 +730,15 @@ MigrateInfo::MigrateInfo(const ShardId& a_to,
const NamespaceString& a_nss,
const UUID& a_uuid,
const BSONObj& a_min,
- const boost::optional<BSONObj>& a_max,
+ const BSONObj& a_max,
const ChunkVersion& a_version,
- const MoveChunkRequest::ForceJumbo a_forceJumbo,
- boost::optional<int64_t> maxChunkSizeBytes)
+ const MoveChunkRequest::ForceJumbo a_forceJumbo)
: nss(a_nss),
uuid(a_uuid),
minKey(a_min),
maxKey(a_max),
version(a_version),
- forceJumbo(a_forceJumbo),
- optMaxChunkSizeBytes(maxChunkSizeBytes) {
+ forceJumbo(a_forceJumbo) {
invariant(a_to.isValid());
invariant(a_from.isValid());
@@ -1135,10 +772,6 @@ 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,
@@ -1222,15 +855,13 @@ DataSizeInfo::DataSizeInfo(const ShardId& shardId,
const ChunkRange& chunkRange,
const ChunkVersion& version,
const KeyPattern& keyPattern,
- bool estimatedValue,
- int64_t maxSize)
+ bool estimatedValue)
: shardId(shardId),
nss(nss),
uuid(uuid),
chunkRange(chunkRange),
version(version),
keyPattern(keyPattern),
- estimatedValue(estimatedValue),
- maxSize(maxSize) {}
+ estimatedValue(estimatedValue) {}
} // namespace mongo
generated by cgit v1.2.3 (git 2.39.1) at 2026-09-11 11:48:18 +0000