Skip to content

Cherrypick recent DD changes from release-7.3 #11754

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.

Sign up for GitHub

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

Merged
kakaiu merged 3 commits into from
Nov 6, 2024
Merged

Cherrypick recent DD changes from release-7.3 #11754

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c81e957
[Release-7.3] TeamRedundant and TeamUnhealthy data moves choose best …
kakaiu Sep 18, 2024
6ac0baf
[Release-7.3] Delay team remover when space pivot is low (#11665)
kakaiu Sep 19, 2024
2237199
[Release-7.3] Validate ServerTeam count per server in simulation (#11…
kakaiu Sep 26, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions fdbclient/ServerKnobs.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -177,6 +177,9 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi

init( ENABLE_REPLICA_CONSISTENCY_CHECK_ON_DATA_MOVEMENT, true );
init( CONSISTENCY_CHECK_REQUIRED_REPLICAS, 1 );

init( PROBABILITY_TEAM_REDUNDANT_DATAMOVE_CHOOSE_TRUE_BEST_DEST, 0.0 ); if (isSimulated) PROBABILITY_TEAM_REDUNDANT_DATAMOVE_CHOOSE_TRUE_BEST_DEST = deterministicRandom()->random01();
init( PROBABILITY_TEAM_UNHEALTHY_DATAMOVE_CHOOSE_TRUE_BEST_DEST, 0.0 ); if (isSimulated) PROBABILITY_TEAM_UNHEALTHY_DATAMOVE_CHOOSE_TRUE_BEST_DEST = deterministicRandom()->random01();

// Data distribution
init( AVAILABLE_SPACE_PIVOT_RATIO, 0.5 );
Expand Down Expand Up @@ -368,6 +371,7 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi
init( DD_MAXIMUM_LARGE_TEAM_CLEANUP, 10000 ); if( randomize && BUGGIFY ) DD_MAXIMUM_LARGE_TEAM_CLEANUP = 10;
init( DD_LARGE_TEAM_DELAY, 60.0 );
init( DD_FIX_WRONG_REPLICAS_DELAY, 60.0 );
init (DD_VALIDATE_SERVER_TEAM_COUNT_AFTER_BUILD_TEAM, false ); if (isSimulated) DD_VALIDATE_SERVER_TEAM_COUNT_AFTER_BUILD_TEAM = true;

// BulkLoading
init( BULKLOAD_FILE_BYTES_MAX, 1*1024*1024*1024 ); // 1GB
Expand All @@ -377,6 +381,7 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi
init( DD_BULKLOAD_SCHEDULE_MIN_INTERVAL_SEC, 2.0 ); if( randomize && BUGGIFY ) DD_BULKLOAD_SCHEDULE_MIN_INTERVAL_SEC = deterministicRandom()->random01() * 10 + 1;

// TeamRemover
init( TR_LOW_SPACE_PIVOT_DELAY_SEC, 0 ); if (isSimulated) TR_LOW_SPACE_PIVOT_DELAY_SEC = deterministicRandom()->randomInt(0, 3);
init( TR_FLAG_DISABLE_MACHINE_TEAM_REMOVER, false ); if( randomize && BUGGIFY ) TR_FLAG_DISABLE_MACHINE_TEAM_REMOVER = deterministicRandom()->random01() < 0.1 ? true : false; // false by default. disable the consistency check when it's true
init( TR_REMOVE_MACHINE_TEAM_DELAY, 60.0 ); if( randomize && BUGGIFY ) TR_REMOVE_MACHINE_TEAM_DELAY = deterministicRandom()->random01() * 60.0;
init( TR_FLAG_REMOVE_MT_WITH_MOST_TEAMS, true ); if( randomize && BUGGIFY ) TR_FLAG_REMOVE_MT_WITH_MOST_TEAMS = deterministicRandom()->random01() < 0.1 ? true : false;
Expand Down
12 changes: 12 additions & 0 deletions fdbclient/include/fdbclient/ServerKnobs.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -201,6 +201,11 @@ class SWIFT_CXX_IMMORTAL_SINGLETON_TYPE ServerKnobs : public KnobsImpl<ServerKno
bool ENABLE_REPLICA_CONSISTENCY_CHECK_ON_DATA_MOVEMENT;
int CONSISTENCY_CHECK_REQUIRED_REPLICAS;

// Probability that a team redundant data move set TrueBest when get destination team
double PROBABILITY_TEAM_REDUNDANT_DATAMOVE_CHOOSE_TRUE_BEST_DEST;
// Probability that a team unhealthy data move set TrueBest when get destination team
double PROBABILITY_TEAM_UNHEALTHY_DATAMOVE_CHOOSE_TRUE_BEST_DEST;

// Data distribution
// DD use AVAILABLE_SPACE_PIVOT_RATIO to calculate pivotAvailableSpaceRatio. Given an array that's descend
// sorted by available space ratio, the pivot position is AVAILABLE_SPACE_PIVOT_RATIO * team count.
Expand Down Expand Up @@ -346,8 +351,15 @@ class SWIFT_CXX_IMMORTAL_SINGLETON_TYPE ServerKnobs : public KnobsImpl<ServerKno
bool DD_ENABLE_REBALANCE_STORAGE_QUEUE_WITH_LIGHT_WRITE_SHARD; // Enable to allow storage queue rebalancer to move
// light-traffic shards out of the overloading server
double DD_WAIT_TSS_DATA_MOVE_DELAY;
bool DD_VALIDATE_SERVER_TEAM_COUNT_AFTER_BUILD_TEAM; // Enable to validate server team count per server after build
// team

// TeamRemover to remove redundant teams
double TR_LOW_SPACE_PIVOT_DELAY_SEC; // teamRedundant data moves can make the min SS available % smaller in
// particular when the majority of SSes have low available %. So, when the
// pivot is below the target, teamRemover wait for the specified time to check
// the pivot again. teamRemover triggers teamRedundant data moves only when the
// pivot is above the target.
bool TR_FLAG_DISABLE_MACHINE_TEAM_REMOVER; // disable the machineTeamRemover actor
double TR_REMOVE_MACHINE_TEAM_DELAY; // wait for the specified time before try to remove next machine team
bool TR_FLAG_REMOVE_MT_WITH_MOST_TEAMS; // guard to select which machineTeamRemover logic to use
Expand Down
80 changes: 52 additions & 28 deletions fdbserver/DDRelocationQueue.actor.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1429,6 +1429,21 @@ void validateBulkLoadRelocateData(const RelocateData& rd, const std::vector<UID>
return;
}

// With probability, set wantTrueBestIfMoveout true for teamUnhealthy data moves and teamRedundant data moves only.
// This flag takes effect in getTeam. When the flag is set true, DD always getBestTeam for teamRedundant data moves and
// do getBestTeam for a teamRedundant data move if the data move decides to move data out of a SS.
bool getWantTrueBestIfMoveout(int priority) {
if (priority == SERVER_KNOBS->PRIORITY_TEAM_UNHEALTHY) {
return deterministicRandom()->random01() <
SERVER_KNOBS->PROBABILITY_TEAM_UNHEALTHY_DATAMOVE_CHOOSE_TRUE_BEST_DEST;
} else if (priority == SERVER_KNOBS->PRIORITY_TEAM_REDUNDANT) {
return deterministicRandom()->random01() <
SERVER_KNOBS->PROBABILITY_TEAM_REDUNDANT_DATAMOVE_CHOOSE_TRUE_BEST_DEST;
} else {
return false;
}
}

// This actor relocates the specified keys to a good place.
// The inFlightActor key range map stores the actor for each RelocateData
ACTOR Future<Void> dataDistributionRelocator(DDQueue* self,
Expand All @@ -1453,6 +1468,7 @@ ACTOR Future<Void> dataDistributionRelocator(DDQueue* self,
state double startTime = now();
state std::vector<UID> destIds;
state WantTrueBest wantTrueBest(isValleyFillerPriority(rd.priority));
state WantTrueBestIfMoveout wantTrueBestIfMoveout(getWantTrueBestIfMoveout(rd.priority));
state uint64_t debugID = deterministicRandom()->randomUInt64();
state bool enableShardMove = SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA && SERVER_KNOBS->ENABLE_DD_PHYSICAL_SHARD;
state bool doBulkLoading = rd.bulkLoadTask.present();
Expand All @@ -1467,6 +1483,7 @@ ACTOR Future<Void> dataDistributionRelocator(DDQueue* self,
.detail("KeyBegin", rd.keys.begin)
.detail("KeyEnd", rd.keys.end)
.detail("Priority", rd.priority)
.detail("WantTrueBestIfMoveout", wantTrueBestIfMoveout)
.detail("SuppressedEventCount", self->suppressIntervals);

if (relocateShardInterval.severity != SevDebug) {
Expand Down Expand Up @@ -1614,6 +1631,7 @@ ACTOR Future<Void> dataDistributionRelocator(DDQueue* self,
req.completeSources = rd.completeSources;
req.storageQueueAware = SERVER_KNOBS->ENABLE_STORAGE_QUEUE_AWARE_TEAM_SELECTION;
req.findTeamForBulkLoad = doBulkLoading;
req.wantTrueBestIfMoveout = wantTrueBestIfMoveout;

if (enableShardMove && tciIndex == 1) {
ASSERT(physicalShardIDCandidate != UID().first() &&
Expand All @@ -1622,8 +1640,8 @@ ACTOR Future<Void> dataDistributionRelocator(DDQueue* self,
self->physicalShardCollection->tryGetAvailableRemoteTeamWith(
physicalShardIDCandidate, metrics, debugID);
if (!remoteTeamWithPhysicalShard.second) {
// Physical shard with `physicalShardIDCandidate` is not available. Retry selecting new
// dst physical shard.
// Physical shard with `physicalShardIDCandidate` is not available. Retry selecting
// new dst physical shard.
self->retryFindDstReasonCount[DDQueue::RetryFindDstReason::NoAvailablePhysicalShard]++;
foundTeams = false;
break;
Expand All @@ -1636,8 +1654,8 @@ ACTOR Future<Void> dataDistributionRelocator(DDQueue* self,
}
}

// bestTeam.second = false if the bestTeam in the teamCollection (in the DC) does not have any
// server that hosts the relocateData. This is possible, for example, in a fearless
// bestTeam.second = false if the bestTeam in the teamCollection (in the DC) does not have
// any server that hosts the relocateData. This is possible, for example, in a fearless
// configuration when the remote DC is just brought up.
Future<std::pair<Optional<Reference<IDataDistributionTeam>>, bool>> fbestTeam =
brokenPromiseToNever(self->teamCollections[tciIndex].getTeam.getReply(req));
Expand Down Expand Up @@ -1680,11 +1698,11 @@ ACTOR Future<Void> dataDistributionRelocator(DDQueue* self,
if (enableShardMove) {
if (tciIndex == 1 && !forceToUseNewPhysicalShard) {
// critical to the correctness of team selection by PhysicalShardCollection
// tryGetAvailableRemoteTeamWith() enforce to select a remote team paired with a primary
// team Thus, tryGetAvailableRemoteTeamWith() may select an almost full remote team In
// this case, we must re-select a remote team We set foundTeams = false to avoid
// finishing team selection Then, forceToUseNewPhysicalShard is set, which enforce to
// use getTeam to select a remote team
// tryGetAvailableRemoteTeamWith() enforce to select a remote team paired with a
// primary team Thus, tryGetAvailableRemoteTeamWith() may select an almost full
// remote team In this case, we must re-select a remote team We set foundTeams =
// false to avoid finishing team selection Then, forceToUseNewPhysicalShard is set,
// which enforce to use getTeam to select a remote team
bool minAvailableSpaceRatio = bestTeam.first.get()->getMinAvailableSpaceRatio(true);
if (minAvailableSpaceRatio < SERVER_KNOBS->TARGET_AVAILABLE_SPACE_RATIO) {
self->retryFindDstReasonCount[DDQueue::RetryFindDstReason::RemoteTeamIsFull]++;
Expand All @@ -1693,11 +1711,11 @@ ACTOR Future<Void> dataDistributionRelocator(DDQueue* self,
}

// critical to the correctness of team selection by PhysicalShardCollection
// tryGetAvailableRemoteTeamWith() enforce to select a remote team paired with a primary
// team Thus, tryGetAvailableRemoteTeamWith() may select an unhealthy remote team In
// this case, we must re-select a remote team We set foundTeams = false to avoid
// finishing team selection Then, forceToUseNewPhysicalShard is set, which enforce to
// use getTeam to select a remote team
// tryGetAvailableRemoteTeamWith() enforce to select a remote team paired with a
// primary team Thus, tryGetAvailableRemoteTeamWith() may select an unhealthy remote
// team In this case, we must re-select a remote team We set foundTeams = false to
// avoid finishing team selection Then, forceToUseNewPhysicalShard is set, which
// enforce to use getTeam to select a remote team
if (!bestTeam.first.get()->isHealthy()) {
self->retryFindDstReasonCount
[DDQueue::RetryFindDstReason::RemoteTeamIsNotHealthy]++;
Expand All @@ -1707,11 +1725,11 @@ ACTOR Future<Void> dataDistributionRelocator(DDQueue* self,
}

bestTeams.emplace_back(bestTeam.first.get(), true);
// Always set bestTeams[i].second = true to disable optimization in data move between DCs
// for the correctness of PhysicalShardCollection
// Currently, enabling the optimization will break the invariant of PhysicalShardCollection
// Invariant: once a physical shard is created with a specific set of SSes, this SS set will
// never get changed.
// Always set bestTeams[i].second = true to disable optimization in data move between
// DCs for the correctness of PhysicalShardCollection Currently, enabling the
// optimization will break the invariant of PhysicalShardCollection Invariant: once a
// physical shard is created with a specific set of SSes, this SS set will never get
// changed.

if (tciIndex == 0) {
ASSERT(foundTeams);
Expand All @@ -1731,9 +1749,9 @@ ACTOR Future<Void> dataDistributionRelocator(DDQueue* self,
self->retryFindDstReasonCount
[DDQueue::RetryFindDstReason::NoAvailablePhysicalShard]++;
if (wantTrueBest) {
// Next retry will likely get the same team, and we know that we can't reuse
// any existing physical shard in this team. So force to create new physical
// shard.
// Next retry will likely get the same team, and we know that we can't
// reuse any existing physical shard in this team. So force to create
// new physical shard.
forceToUseNewPhysicalShard = true;
}
foundTeams = false;
Expand Down Expand Up @@ -1784,6 +1802,9 @@ ACTOR Future<Void> dataDistributionRelocator(DDQueue* self,
TraceEvent(destOverloadedCount > 50 ? SevInfo : SevDebug, "DestSSBusy", distributorId)
.suppressFor(1.0)
.detail("TraceID", rd.randomId)
.detail("WantTrueBestIfMoveout", wantTrueBestIfMoveout)
.detail("IsRestore", rd.isRestore())
.detail("Priority", rd.priority)
.detail("StuckCount", stuckCount)
.detail("DestOverloadedCount", destOverloadedCount)
.detail("TeamCollectionId", tciIndex)
Expand All @@ -1802,6 +1823,9 @@ ACTOR Future<Void> dataDistributionRelocator(DDQueue* self,
TraceEvent(stuckCount > 50 ? SevWarnAlways : SevWarn, "BestTeamStuck", distributorId)
.suppressFor(1.0)
.detail("TraceID", rd.randomId)
.detail("WantTrueBestIfMoveout", wantTrueBestIfMoveout)
.detail("IsRestore", rd.isRestore())
.detail("Priority", rd.priority)
.detail("StuckCount", stuckCount)
.detail("DestOverloadedCount", destOverloadedCount)
.detail("TeamCollectionId", tciIndex)
Expand Down Expand Up @@ -2141,8 +2165,8 @@ ACTOR Future<Void> dataDistributionRelocator(DDQueue* self,
auto serverIds = bestTeams[i].first->getServerIDs();
selectedTeams.push_back(ShardsAffectedByTeamFailure::Team(serverIds, i == 0));
}
// The update of PhysicalShardToTeams, PhysicalShardInstances, keyRangePhysicalShardIDMap should
// be atomic
// The update of PhysicalShardToTeams, PhysicalShardInstances, keyRangePhysicalShardIDMap
// should be atomic
self->physicalShardCollection->updatePhysicalShardCollection(
rd.keys, rd.isRestore(), selectedTeams, rd.dataMoveId.first(), metrics, debugID);
}
Expand Down Expand Up @@ -2242,8 +2266,8 @@ inline double getWorstCpu(const HealthMetrics& metrics, const std::vector<UID>&
return cpu;
}

// Move the shard with the top K highest read density of sourceTeam's to destTeam if sourceTeam has much more read load
// than destTeam
// Move the shard with the top K highest read density of sourceTeam's to destTeam if sourceTeam has much more read
// load than destTeam
ACTOR Future<bool> rebalanceReadLoad(DDQueue* self,
DataMovementReason moveReason,
Reference<IDataDistributionTeam> sourceTeam,
Expand Down Expand Up @@ -2785,8 +2809,8 @@ struct DDQueueImpl {
}
}
} catch (Error& e) {
if (e.code() != error_code_broken_promise && // FIXME: Get rid of these broken_promise errors every time we
// are killed by the master dying
if (e.code() != error_code_broken_promise && // FIXME: Get rid of these broken_promise errors every time
// we are killed by the master dying
e.code() != error_code_movekeys_conflict && e.code() != error_code_data_move_cancelled &&
e.code() != error_code_data_move_dest_team_not_found)
TraceEvent(SevError, "DataDistributionQueueError", self->distributorId).error(e);
Expand Down
12 changes: 6 additions & 6 deletions fdbserver/DDShardTracker.actor.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -217,14 +217,14 @@ ACTOR Future<Void> shardUsableRegions(DataDistributionTracker::SafeAccessor self
double expectedCompletionSeconds = self()->shards->size() * 1.0 / SERVER_KNOBS->DD_SHARD_USABLE_REGION_CHECK_RATE;
double delayTime = deterministicRandom()->random01() * expectedCompletionSeconds;
wait(delayJittered(delayTime));
auto [destTeams, srcTeams] = self()->shardsAffectedByTeamFailure->getTeamsForFirstShard(keys);
if (destTeams.size() < self()->usableRegions) {
auto [newTeam, previousTeam] = self()->shardsAffectedByTeamFailure->getTeamsForFirstShard(keys);
if (newTeam.size() < self()->usableRegions) {
TraceEvent(SevWarn, "ShardUsableRegionMismatch", self()->distributorId)
.suppressFor(5.0)
.detail("DestTeamSize", destTeams.size())
.detail("SrcTeamSize", srcTeams.size())
.detail("DestServers", describe(destTeams))
.detail("SrcServers", describe(srcTeams))
.detail("NewTeamSize", newTeam.size())
.detail("PreviousTeamSize", previousTeam.size())
.detail("NewServers", describe(newTeam))
.detail("PreviousServers", describe(previousTeam))
.detail("UsableRegion", self()->usableRegions)
.detail("Shard", keys);
RelocateShard rs(keys, DataMovementReason::POPULATE_REGION, RelocateReason::OTHER);
Expand Down
Loading