Skip to content

Make ShardsAffectedByTeamFailure::moveShard range-safe #13787

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

Draft
saintstack wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

Make ShardsAffectedByTeamFailure::moveShard range-safe #13787

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
2 changes: 2 additions & 0 deletions fdbserver/core/ServerKnobs.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -285,6 +285,8 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi
// guess (thus 20 moves). Do not buggify this too small: testing under artificial scarcity results in
// uninteresting degenerate cases.
init( DD_MAX_PIPELINE_MOVES, 1000 ); if( randomize && buggify() ) DD_MAX_PIPELINE_MOVES = 20;
init( DD_RECONCILE_SHARDS_ON_EXCLUDE, true );
init( DD_SPLIT_TRACKED_SHARDS_ON_MOVE, true );
init( DD_REBALANCE_RESET_AMOUNT, 30 );
init( INFLIGHT_PENALTY_HEALTHY, 1.0 );
init( INFLIGHT_PENALTY_UNHEALTHY, 500.0 );
Expand Down
12 changes: 12 additions & 0 deletions fdbserver/core/include/fdbserver/core/Knobs.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -160,6 +160,18 @@ class SWIFT_CXX_IMMORTAL_SINGLETON_TYPE ServerKnobs : public KnobsImpl<ServerKno
int DD_QUEUE_MAX_KEY_SERVERS;
int DD_REBALANCE_PARALLELISM;
int DD_MAX_PIPELINE_MOVES; // Hard cap on total relocations DD tracks (queued + in-flight).
bool DD_RECONCILE_SHARDS_ON_EXCLUDE; // On graceful exclude, if the server's on-disk serverKeys are empty
// (canRemove) but its in-memory ShardsAffectedByTeamFailure count is
// still > 0, reconcile the map to the on-disk truth so removal is not
// blocked. When false, fall back to the legacy behavior of waiting for
// the count to reach 0 on its own.
bool DD_SPLIT_TRACKED_SHARDS_ON_MOVE; // Have ShardsAffectedByTeamFailure::moveShard() split the tracked
// shard at the move's key boundaries, so that a move of a strict
// sub-range of a tracked shard erases the source teams for the range
// that actually moved. When false, restore the legacy behavior of
// appending the destination team to the whole enclosing shard
// without erasing the source, which permanently over-counts drained
// source servers in storageServerShards.
int DD_REBALANCE_RESET_AMOUNT;
double INFLIGHT_PENALTY_HEALTHY;
double INFLIGHT_PENALTY_REDUNDANT;
Expand Down
62 changes: 59 additions & 3 deletions fdbserver/datadistributor/DDTxnProcessor.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -853,13 +853,69 @@ class DDTxnProcessorImpl {
if (ver > addedVersion + SERVER_KNOBS->MAX_READ_TRANSACTION_LIFE_VERSIONS) {
bool canRemove = co_await canRemoveStorageServer(tr, serverID);
auto shards = shardsAffectedByTeamFailure->getNumberOfShards(serverID);
TraceEvent(SevVerbose, "WaitForAllDataRemoved")
// Polled every ALL_DATA_REMOVED_DELAY for every server under removal, so it is
// suppressed; the diagnostic signal is ShardsAffectedStrandedOnExclude below.
TraceEvent(SevInfo, "WaitForAllDataRemoved")
.suppressFor(30.0)
.detail("Server", serverID)
.detail("CanRemove", canRemove)
.detail("Shards", shards);
ASSERT_GE(shards, 0);
if (canRemove && shards == 0) {
co_return;
if (canRemove) {
if (shards == 0) {
co_return;
}
// canRemoveStorageServer() consults the authoritative on-disk serverKeys: when it is
// true the server owns no data, and DDTeamCollection.actor.cpp:5964 documents the
// matching in-memory invariant as a commented-out ASSERT -- no team containing the
// server should remain in ShardsAffectedByTeamFailure. Nothing enforces it. The count
// can be stranded > 0 by any move whose key range only partially covers a tracked
// range: moveShard()'s partial-overlap branch appends the destination team without
// erasing the drained source team, and only a later fully-covering move, a
// defineShard(), or a DD restart rewrites that entry. Such a stale count blocks
// removeStorageServer() indefinitely. Since the on-disk state is authoritative and
// already says the server is empty, reconcile the map to it and trace loudly rather
// than hang. moveShard() is range-safe as of DD_SPLIT_TRACKED_SHARDS_ON_MOVE; this
// stays as enforcement, because we cannot prove every source of partial-range moves
// has been found.
//
// This event repeating is the "still stranded" signal; it is the one trace that makes
// this failure mode diagnosable from logs alone.
TraceEvent(SevWarnAlways, "ShardsAffectedStrandedOnExclude")
.suppressFor(60.0)
.detail("Server", serverID)
.detail("StrandedShards", shards)
.detail("TotalTrackedShards", shardsAffectedByTeamFailure->getNumberOfShards())
.detail("WillReconcile", SERVER_KNOBS->DD_RECONCILE_SHARDS_ON_EXCLUDE);
if (SERVER_KNOBS->DD_RECONCILE_SHARDS_ON_EXCLUDE) {
CODE_PROBE(true,
"Reconciled a stranded ShardsAffectedByTeamFailure count on graceful "
"exclude",
probe::decoration::rare);
double start = now();
auto scrub = shardsAffectedByTeamFailure->scrubServer(serverID);
TraceEvent(SevWarnAlways, "ShardsAffectedReconciledOnExclude")
.detail("Server", serverID)
.detail("StrandedShards", shards)
.detail("ShardsScanned", scrub.shardsScanned)
.detail("ShardsRewritten", scrub.shardsRewritten)
.detail("OwnerlessShards", scrub.ownerlessShards)
.detail("SampleRange", scrub.sampleRange)
.detail("Elapsed", now() - start);
auto remaining = shardsAffectedByTeamFailure->getNumberOfShards(serverID);
ASSERT_WE_THINK(remaining == 0);
if (remaining == 0) {
co_return;
}
// Unreachable by construction: scrubServer() drops every team naming the server,
// and only those teams contribute to the count. Rather than deregister a server
// the map still references, keep polling and make the anomaly loud.
TraceEvent(SevWarnAlways, "ShardsAffectedReconcileIncomplete")
.detail("Server", serverID)
.detail("RemainingShards", remaining);
}
// Knob disabled: fall through to the delay below and keep waiting for the count to
// reach 0 on its own (legacy behavior).
}
}

Expand Down
152 changes: 127 additions & 25 deletions fdbserver/datadistributor/ShardsAffectedByTeamFailure.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
*/

#include "ShardsAffectedByTeamFailure.h"
#include "fdbserver/core/Knobs.h"

std::vector<KeyRange> ShardsAffectedByTeamFailure::getShardsFor(Team team) const {
std::vector<KeyRange> r;
Expand Down Expand Up @@ -116,16 +117,72 @@ void ShardsAffectedByTeamFailure::defineShard(KeyRangeRef keys) {
check();
}

void ShardsAffectedByTeamFailure::splitTrackedShardsAtBoundaries(KeyRangeRef keys) {
// Recreate only the boundary points of `keys`. defineShard() must not be used here: it would merge all
// tracked shards inside keys into one entry, losing the distinct destination teams of overlapping newer
// moves. team_shards is keyed by the EXACT (team, range) pair, so a tracked shard that straddles a
// boundary has to be removed from it before the split and its pieces re-added afterwards, or the
// team_shards <-> shard_teams invariant (see check()) breaks.
std::vector<KeyRange> rangesToSplit;
auto beginRange = shard_teams.rangeContaining(keys.begin);
if (beginRange->begin() != keys.begin) {
rangesToSplit.push_back(beginRange->range());
}
auto endRange = shard_teams.rangeContaining(keys.end);
if (endRange->begin() != keys.end && (rangesToSplit.empty() || rangesToSplit.back() != endRange->range())) {
rangesToSplit.push_back(endRange->range());
}
if (rangesToSplit.empty()) {
// Both boundaries already exist; modify() would be a no-op.
return;
}
for (const auto& range : rangesToSplit) {
for (const auto& team : shard_teams.rangeContaining(range.begin)->value().first) {
erase(team, range);
}
}
shard_teams.modify(keys);
for (const auto& range : rangesToSplit) {
for (auto splitRange : shard_teams.containedRanges(range)) {
for (const auto& team : splitRange.value().first) {
insert(team, splitRange.range());
}
}
}
}

void ShardsAffectedByTeamFailure::moveShard(KeyRangeRef keys, std::vector<Team> destinationTeams) {
/*TraceEvent("ShardsAffectedByTeamFailureMove")
.detail("KeyBegin", keys.begin)
.detail("KeyEnd", keys.end)
.detail("NewTeamSize", destinationTeam.size())
.detail("NewTeam", describe(destinationTeam));*/

// A relocation is frequently launched for a STRICT SUB-RANGE of a tracked shard, for three independent
// reasons. (1) A relocation's keys are captured when it is created, so an intervening defineShard() --
// notably from a shard merge, which coarsens several tracked shards into one before sending its own
// relocation -- can widen the tracked shard underneath a request that is still queued. (2) DDQueue
// truncates already-queued relocations against overlapping newer ones (queueRelocation). (3)
// launchQueuedWork starts a relocator per range returned by getRangesAffectedByInsertion, which includes
// the pieces of a live relocator straddling the new range's boundaries, i.e. strict subsets of an older
// relocation's range. Without a boundary split such a move hits the partial-overlap branch below, which
// appends the destination team but never erases the drained source team -- so a source server stays counted
// in storageServerShards forever, and a gracefully excluded server is never removable because its removal
// gate requires getNumberOfShards(server) == 0. Splitting the tracked shard at the move's boundaries first
// makes every affected shard fully contained, so the source teams are erased for exactly the range that
// actually moved. The extra boundaries are transient: the next defineShard() from a shard-tracker
// split/merge over a coarser range merges them back.
if (SERVER_KNOBS->DD_SPLIT_TRACKED_SHARDS_ON_MOVE) {
splitTrackedShardsAtBoundaries(keys);
}

auto ranges = shard_teams.intersectingRanges(keys);
std::vector<std::pair<std::pair<std::vector<Team>, std::vector<Team>>, KeyRange>> modifiedShards;
for (auto it = ranges.begin(); it != ranges.end(); ++it) {
// After splitTrackedShardsAtBoundaries() every intersecting range is contained in keys, so the
// partial-overlap branch below is unreachable. It is retained as a safety net rather than an assert
// because losing a destination team in production is worse than over-counting a source team.
ASSERT_WE_THINK(!SERVER_KNOBS->DD_SPLIT_TRACKED_SHARDS_ON_MOVE || keys.contains(it->range()));
if (keys.contains(it->range())) {
// erase the many teams that were associated with this one shard
for (auto t = it->value().first.begin(); t != it->value().first.end(); ++t) {
Expand Down Expand Up @@ -170,31 +227,9 @@ std::vector<KeyRange> ShardsAffectedByTeamFailure::cancelMove(KeyRangeRef keys,
const std::vector<Team>& destinationTeams,
const std::vector<Team>& sourceTeams) {
std::vector<KeyRange> restoredRanges;
// A later shard split or merge can leave the cancelled move range strictly inside a tracked shard. Recreate only
// the move's boundary points before removing its destinations. defineShard() would merge all tracked shards inside
// keys, losing the distinct destinations of overlapping newer moves.
std::vector<KeyRange> rangesToSplit;
auto beginRange = shard_teams.rangeContaining(keys.begin);
if (beginRange->begin() != keys.begin) {
rangesToSplit.push_back(beginRange->range());
}
auto endRange = shard_teams.rangeContaining(keys.end);
if (endRange->begin() != keys.end && (rangesToSplit.empty() || rangesToSplit.back() != endRange->range())) {
rangesToSplit.push_back(endRange->range());
}
for (const auto& range : rangesToSplit) {
for (const auto& team : shard_teams.rangeContaining(range.begin)->value().first) {
erase(team, range);
}
}
shard_teams.modify(keys);
for (const auto& range : rangesToSplit) {
for (auto splitRange : shard_teams.containedRanges(range)) {
for (const auto& team : splitRange.value().first) {
insert(team, splitRange.range());
}
}
}
// A later shard split or merge can leave the cancelled move range strictly inside a tracked shard, so
// recreate the move's boundary points before removing its destinations.
splitTrackedShardsAtBoundaries(keys);
auto ranges = shard_teams.containedRanges(keys);
for (auto it = ranges.begin(); it != ranges.end(); ++it) {
std::vector<Team> retainedTeams;
Expand Down Expand Up @@ -311,6 +346,73 @@ void ShardsAffectedByTeamFailure::removeFailedServerForRange(KeyRangeRef keys, c
check();
}

ShardsAffectedByTeamFailure::ScrubResult ShardsAffectedByTeamFailure::scrubServer(const UID& serverID) {
auto containsServer = [&serverID](const std::vector<Team>& teams) {
return std::any_of(teams.begin(), teams.end(), [&serverID](const Team& t) { return t.hasServer(serverID); });
};
auto without = [&serverID](const std::vector<Team>& teams) {
std::vector<Team> retained;
for (const auto& t : teams) {
if (!t.hasServer(serverID)) {
retained.push_back(t);
}
}
return retained;
};

ScrubResult result;
std::vector<KeyRange> restartRanges;
// Values are mutated in place; no shard_teams range is inserted or erased, so the iteration stays valid.
auto rs = shard_teams.ranges();
for (auto it = rs.begin(); it != rs.end(); ++it) {
++result.shardsScanned;
auto& teams = it->value();
const bool inCurrent = containsServer(teams.first);
if (!inCurrent && !containsServer(teams.second)) {
continue;
}

const KeyRange range = it->range();
std::vector<Team> retained = without(teams.first);
std::vector<Team> retainedPrev = without(teams.second);

if (inCurrent) {
// Only the current-team list is mirrored in team_shards (and therefore in
// storageServerShards), so it is the only one that needs the erase/insert dance.
for (const auto& t : teams.first) {
erase(t, range);
}
if (retained.empty()) {
retained = retainedPrev;
retainedPrev.clear();
}
for (const auto& t : retained) {
insert(t, range);
}
teams.first = retained;
if (retained.empty()) {
++result.ownerlessShards;
restartRanges.push_back(range);
}
}
// getSourceServerIdsFor() prefers the previous-source list, so a drained server left there would
// still be handed out as a relocation source.
teams.second = retainedPrev;

if (result.shardsRewritten == 0) {
result.sampleRange = range;
}
++result.shardsRewritten;
}
check();

// Sent after the map is consistent: the receiver re-enters this object via defineShard().
for (const auto& range : restartRanges) {
restartShardTracker.send(range);
}
return result;
}

auto ShardsAffectedByTeamFailure::intersectingRanges(KeyRangeRef keyRange) const -> decltype(shard_teams)::ConstRanges {
return shard_teams.intersectingRanges(keyRange);
}
Expand Down
41 changes: 37 additions & 4 deletions fdbserver/datadistributor/ShardsAffectedByTeamFailure.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,14 @@ class ShardsAffectedByTeamFailure : public ReferenceCounted<ShardsAffectedByTeam
std::string toString() const { return describe(servers); };
};

// Outcome of scrubServer(), for tracing.
struct ScrubResult {
int shardsScanned = 0; // shard_teams entries examined
int shardsRewritten = 0; // shard_teams entries whose team lists changed
int ownerlessShards = 0; // entries left with an empty current-team list (see scrubServer)
KeyRange sampleRange; // one rewritten range, for diagnosis
};

// This tracks the data distribution on the data distribution server so that teamTrackers can
// relocate the right shards when a team is degraded.

Expand All @@ -72,10 +80,13 @@ class ShardsAffectedByTeamFailure : public ReferenceCounted<ShardsAffectedByTeam
// of what servers correspond to each shard is a copy or union of the shards already there
// - The teams associated with each shard reflect either the sources for non-moving shards
// or the destination team for in-flight shards (the change is atomic with respect to team selection).
// moveShard() changes the servers associated with a shard and will never adjust the shard
// boundaries. If a move is received for a shard that has been redefined (the exact shard is
// no longer in the map), the servers will be set for all contained shards and added to all
// intersecting shards.
// moveShard() changes the servers associated with a shard. It splits the tracked shard at the move's
// boundaries first (see DD_SPLIT_TRACKED_SHARDS_ON_MOVE) so that a move of a strict sub-range of a
// tracked shard still erases the source teams for exactly the range that moved; the added boundaries
// are merged away by the next defineShard() over a coarser range. Historically moveShard() never
// adjusted boundaries, and a sub-range move instead added the destination to the whole enclosing
// shard WITHOUT erasing the source -- which permanently over-counted the drained source servers in
// storageServerShards and could block storage server removal indefinitely.

int getNumberOfShards(UID ssID) const;
int getNumberOfShards(Team team) const;
Expand Down Expand Up @@ -136,13 +147,35 @@ class ShardsAffectedByTeamFailure : public ReferenceCounted<ShardsAffectedByTeam

bool removeFailedServerForSingleRange(ShardsAffectedByTeamFailure::Team& team, const UID& id, KeyRangeRef keys);

// Force tracked-shard boundaries at keys.begin and keys.end, leaving the teams attributed to every key
// unchanged. Used before a move or a move cancellation so the operation acts on fully-contained shards.
void splitTrackedShardsAtBoundaries(KeyRangeRef keys);

public:
// return the iterator that traversing all ranges
auto getAllRanges() const -> decltype(shard_teams)::ConstRanges;
auto intersectingRanges(KeyRangeRef keyRange) const -> decltype(shard_teams)::ConstRanges;
// get total shards count
size_t getNumberOfShards() const;
void removeFailedServerForRange(KeyRangeRef keys, const UID& serverID);

// Reconcile the map to the authoritative on-disk fact that `serverID` stores nothing: drop every
// reference to a team that contains it, from both the current-team and previous-source lists.
// getNumberOfShards(serverID) is 0 on return.
//
// This deliberately removes the whole TEAM REFERENCE rather than editing team membership in place the
// way removeFailedServerForRange() does. Editing membership would leave behind a shrunken team (e.g. a
// 2-server team) that exists nowhere in DDTeamCollection -- DDTeamCollection::removeServer() deletes
// the teams containing a removed server rather than shrinking them -- and a shard attributed to a team
// that no team tracker owns is re-issued forever at PRIORITY_TEAM_REDUNDANT by
// DDTeamCollection::teamTracker()'s "team not found" path. In other words, editing membership would
// trade one accounting bug for a self-sustaining relocation loop.
//
// Where dropping the reference would leave a shard with no current team, the previous-source teams are
// promoted (the same fallback cancelMove() uses); if those are empty too the entry is left with an
// empty current-team list -- the state a freshly-initialized map is in -- and the shard tracker is
// asked to restart over that range. Those cases are counted in ScrubResult::ownerlessShards.
ScrubResult scrubServer(const UID& serverID);
};

#endif // FOUNDATIONDB_SHARDSAFFECTEDBYTEAMFAILURE_H
Loading
Loading