bitflicker64 commented on code in PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#discussion_r4304748201
##########
hugegraph-pd/docs/configuration.md:
##########
@@ -196,6 +202,21 @@ pd:
- Production: `1800` (30 minutes) to reduce overhead
- Large clusters: `3600` (1 hour)
- `pd.initial-store-count`: Set to expected initial store count (e.g., `3` for
3 stores)
+- `pd.balance-leader-enable`: After a Store restarts, its raft groups keep the
leaders they elected
+ on the surviving Stores. Enable this so PD moves leaders back without a
manual
+ `balanceLeaders` call. A check is skipped while any shard group is not in
the `Normal` state,
+ has sent no partition heartbeat in the last 30 seconds, or has a replica
that is not a Leader
+ or Follower (a membership change in progress); while a Store that holds a
shard is not sending
+ heartbeats; and while a split, move or shard balance task is running. A
planned transfer is
Review Comment:
Minor: "while a Store that holds a shard is not sending heartbeats" reads as
if one missed heartbeat stops the check. The code checks the active-store list,
and a Store stays on it for `store.keepAlive-timeout` (300 s by default) after
its last heartbeat, as the PR description says.
Please write "while a Store that holds a shard is not in the active-store
list, which it leaves `store.keepAlive-timeout` after its last heartbeat".
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java:
##########
@@ -582,6 +659,172 @@ public synchronized Map<Integer, Long>
balancePartitionLeader(boolean immediatel
return results;
}
+ /**
+ * The periodic entry of leader balancing. Calls balancePartitionLeader
only when the
+ * leader count per store has drifted further apart than
pd.balance-leader-max-skew, so a
+ * balanced cluster sees no transfers. Skipped while a shard group is not
Normal, has no
+ * recent partition heartbeat or has a shard that is neither Leader nor
Follower, or while a
+ * store that holds a shard is not active, because the target distribution
would include a
+ * replica that cannot take a leader. The plan uses the groups checked
here; a group keeps
+ * its leader when this PD is no longer the leader, the group changed, or
the target trails
+ * the leader by the time the transfer would be sent.
+ *
+ * @return shard group id -> the store id its leader was moved to; empty
when skipped
+ */
+ public synchronized Map<Integer, Long> autoBalancePartitionLeader() throws
PDException {
+ // init() does not schedule this with a negative skew, which would
balance every run
+ int maxSkew = pdConfig.getBalanceLeaderMaxSkew();
+ if (!isLeader() || maxSkew < 0) {
+ lastLeaderIndexReports.clear();
+ return new HashMap<>();
+ }
+
+ List<Metapb.ShardGroup> shardGroups = storeService.getShardGroups();
+ if (shardGroups.isEmpty()) {
+ return new HashMap<>();
+ }
+
+ List<Metapb.Store> stores = storeService.getActiveStores();
+ Set<Long> activeStores = stores.stream()
+ .map(Metapb.Store::getId)
+ .collect(Collectors.toSet());
+ // shard group id -> store id -> the applied index from the store's
last heartbeat
+ Map<Integer, Map<Long, IndexReport>> indexReports = new HashMap<>();
+ for (Metapb.Store store : stores) {
+ for (Metapb.RaftStats raftStats :
store.getStats().getRaftStatsList()) {
+ indexReports.computeIfAbsent(raftStats.getPartitionId(), id ->
new HashMap<>())
+ .put(store.getId(), new IndexReport(store.getId(),
+
raftStats.getCommittedIndex(),
+
store.getLastHeartbeat()));
+ }
+ }
+ // Plan with the leader reports of the previous run and keep this
run's for the next,
+ // including runs that skip below, so a later run can tell the
leader's write rate
+ Map<Integer, IndexReport> previousLeaderReports = new
HashMap<>(lastLeaderIndexReports);
+ lastLeaderIndexReports.clear();
+ for (Metapb.ShardGroup group : shardGroups) {
+ Long leaderStore = getLeaderStore(group);
+ IndexReport report = leaderStore == null ? null :
+ indexReports.getOrDefault(group.getId(),
Map.of())
+ .get(leaderStore);
+ if (report != null) {
+ lastLeaderIndexReports.put(group.getId(), report);
+ }
+ }
+ long now = System.currentTimeMillis();
+ Map<Long, Integer> leaderCount = new HashMap<>();
+ for (Metapb.ShardGroup group : shardGroups) {
+ // A store stays active for keepAlive-timeout after it stops, but
partition
+ // heartbeats from the surviving replicas already mark its groups
Warn.
+ // balancePartitionLeader plans over all groups, so skip the whole
run.
+ if (group.getState() != Metapb.PartitionState.PState_Normal) {
+ log.info("periodic leader balance skipped, shard group {} is
{}",
+ group.getId(), group.getState());
+ return new HashMap<>();
+ }
+ // A group that lost its quorum stops sending stats and keeps its
last state, so
+ // the state above only counts while the group's leader is still
reporting it
+ Metapb.PartitionStats stats =
partitionService.getPartitionStats("", group.getId());
+ if (stats == null || now - stats.getTimestamp() >
BALANCE_LEADER_STATS_MAX_AGE) {
+ log.info("periodic leader balance skipped, shard group {} has
no partition " +
+ "heartbeat in the last {} ms", group.getId(),
+ BALANCE_LEADER_STATS_MAX_AGE);
+ return new HashMap<>();
+ }
+ for (Metapb.Shard shard : group.getShardsList()) {
+ // reallocShards adds a shard with role None while the
membership change runs;
+ // balancePartitionLeader would pick it, or a Learner, as a
transfer target
+ if (shard.getRole() != Metapb.ShardRole.Leader &&
+ shard.getRole() != Metapb.ShardRole.Follower) {
+ log.info("periodic leader balance skipped, shard group {}
has store {} " +
+ "as {}", group.getId(), shard.getStoreId(),
shard.getRole());
+ return new HashMap<>();
+ }
+ if (!activeStores.contains(shard.getStoreId())) {
+ log.info("periodic leader balance skipped, store {} of
shard group {} is " +
+ "not active", shard.getStoreId(), group.getId());
+ return new HashMap<>();
+ }
+ int leader = shard.getRole() == Metapb.ShardRole.Leader ? 1 :
0;
+ leaderCount.merge(shard.getStoreId(), leader, Integer::sum);
+ }
+ }
+
+ int skew = Collections.max(leaderCount.values()) -
Collections.min(leaderCount.values());
+ if (skew <= maxSkew) {
+ log.debug("periodic leader balance skipped, leader count by store:
{}, skew {}",
+ leaderCount, skew);
+ return new HashMap<>();
+ }
+
+ log.info("leader count by store: {}, skew {} exceeds {}, balancing
leaders",
+ leaderCount, skew, maxSkew);
+ return balancePartitionLeader(false, shardGroups, (group, storeId) ->
+ canTakeLeader(group, storeId, indexReports,
previousLeaderReports));
+ }
+
+ private static Long getLeaderStore(Metapb.ShardGroup group) {
+ return group.getShardsList().stream()
+ .filter(shard -> shard.getRole() ==
Metapb.ShardRole.Leader)
+ .map(Metapb.Shard::getStoreId)
+ .findFirst().orElse(null);
+ }
+
+ /**
+ * Checks a periodic transfer right before it is sent: this PD must still
be the leader, the
+ * group must be unchanged since autoBalancePartitionLeader checked it,
and the target's
+ * applied index must trail the leader's by at most
BALANCE_LEADER_MAX_INDEX_GAP. That gap is
+ * estimated from store heartbeats taken at different times, with the
leader's progress in
+ * between taken as its average rate since the previous run, so it is not
a strict bound:
+ * a burst late in that time makes a caught-up target look behind, and a
burst early in it
+ * hides a target that fell behind.
+ */
+ private boolean canTakeLeader(Metapb.ShardGroup group, long storeId,
+ Map<Integer, Map<Long, IndexReport>>
indexReports,
+ Map<Integer, IndexReport>
previousLeaderReports) throws
+
PDException {
+ // A PD that lost leadership during the run must not send what it
planned as leader
+ if (!isLeader()) {
+ log.info("periodic leader balance keeps the leader of shard group
{}, this PD is " +
+ "no longer the leader", group.getId());
+ return false;
+ }
+ if (!group.equals(storeService.getShardGroup(group.getId()))) {
+ log.info("periodic leader balance keeps the leader of shard group
{}, the group " +
+ "changed after the check", group.getId());
+ return false;
+ }
+ Long leaderStore = getLeaderStore(group);
+ Map<Long, IndexReport> reports =
indexReports.getOrDefault(group.getId(), Map.of());
+ IndexReport leader = leaderStore == null ? null :
reports.get(leaderStore);
+ IndexReport target = reports.get(storeId);
+ if (leader == null || target == null) {
+ log.info("periodic leader balance keeps the leader of shard group
{}, no applied " +
+ "index reported by store {} or leader store {}",
group.getId(), storeId,
+ leaderStore);
+ return false;
+ }
+ // Stores send their heartbeats at different times, so a caught-up
target reporting
+ // earlier than the leader looks behind by the writes in between.
Estimate the leader's
+ // index at the time of the target's report, at the average rate it
grew since the
+ // previous run. Without that rate, compare the indexes as reported.
+ long leaderIndex = leader.index();
+ IndexReport previous = previousLeaderReports.get(group.getId());
+ if (previous != null && previous.storeId() == leader.storeId() &&
+ previous.time() < leader.time() && previous.index() <=
leader.index()) {
+ double rate = (double) (leader.index() - previous.index()) /
+ (leader.time() - previous.time());
+ leaderIndex += Math.round(rate * (target.time() - leader.time()));
Review Comment:
Important: Nothing limits the age of the target report this line backdates
to. The description accepts a follower that fell behind after its last report,
which is at most one 30 s heartbeat. A Store stays active for
`store.keepAlive-timeout`, 300 s by default. If its heartbeats stop reaching PD
while its raft peer is alive but behind (an overloaded Store, say), its groups
stay `Normal`, and a minutes-old report passes. Without history, the same
report is deferred. The leader then rejects writes until JRaft ends the
transfer (`electionTimeoutMs`, 3000), which the 1000-entry check was added to
prevent.
Temporary test at b471ea8a: after `warmUp(service, 70_000, 300_000)`,
leaders at 100000 now, store 3 at 75000 reported 250 s ago. With history:
`{2=2, 3=2, 4=3, 5=3}`. Without: `{2=2, 3=2}`.
Please skip a target whose report is older than about 60 s (two store
heartbeats), or cap the backdating at 30 s, and add the case to the lag tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
