arshilkxwork commented on code in PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#discussion_r4303678270
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java:
##########
@@ -582,6 +626,113 @@ 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 that
+ * changes before its transfer is sent, or whose target trails the leader,
keeps its leader.
+ *
+ * @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) {
+ return new HashMap<>();
+ }
+
+ List<Metapb.ShardGroup> shardGroups = storeService.getShardGroups();
+ if (shardGroups.isEmpty()) {
+ return new HashMap<>();
+ }
+
+ Set<Long> activeStores = storeService.getActiveStores().stream()
+ .map(Metapb.Store::getId)
+ .collect(Collectors.toSet());
+ 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, true);
+ }
+
+ /**
+ * Checks a periodic transfer right before it is sent: the group must be
unchanged since
+ * autoBalancePartitionLeader checked it, and the applied index the
target's store last
+ * reported must trail the leader's by at most
BALANCE_LEADER_MAX_INDEX_GAP.
+ */
+ private boolean canTakeLeader(Metapb.ShardGroup group, long storeId,
+ Map<Integer, Map<Long, Long>>
committedIndexMap) throws
+
PDException {
+ 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 = group.getShardsList().stream()
+ .filter(shard -> shard.getRole() ==
Metapb.ShardRole.Leader)
+ .map(Metapb.Shard::getStoreId)
+ .findFirst().orElse(null);
+ Map<Long, Long> indexes =
committedIndexMap.getOrDefault(group.getId(), Map.of());
+ Long leaderIndex = leaderStore == null ? null :
indexes.get(leaderStore);
+ Long targetIndex = indexes.get(storeId);
+ if (leaderIndex == null || targetIndex == null ||
+ leaderIndex - targetIndex > BALANCE_LEADER_MAX_INDEX_GAP) {
Review Comment:
A correction to my reply above: "the first run after this PD becomes leader"
was too strong. The history is keyed by group and leader Store, not by
leadership term, and only a run that finds this PD is not the leader clears it.
If PD leadership is lost and regained, or the group's leader moves away and
back, between two runs, the old entry stays and the rate is averaged over that
whole time. 67b2dc2da says so in the code and docs.
--
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]
