bitflicker64 commented on code in PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#discussion_r4302468235
##########
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:
Important: The 1000-entry gap compares two applied indexes taken at
different times, and the time difference is the same on every run, so under
steady writes a follower that is caught up can be deferred on every run, not
once. `getCommittedIndexStats()` reads the `RaftStats` from each Store's last
store heartbeat. Stores send that heartbeat every 30 s with a fixed delay
(`HgStoreEngineOptions.storeHBInterval = 30`,
`HeartbeatService.doStoreHeartbeat`), and the default
`pd.balance-leader-interval` of 300 s is a multiple of 30 s. When the target
Store's last report is d seconds older than the leader Store's, a follower with
no real lag still looks about (write rate x d) entries behind: at 100 entries/s
on a partition and d = 15 s that is 1500, so the transfer is skipped. Both
heartbeat timers and the 300 s timer drift only by RPC and run time, so the
same pair sees the same offset run after run, and a restarted Store can get
none of its leaders back for as long as the load lasts. Th
at is the case this feature is meant to fix, and the new tests only use
indexes reported at the same moment.
Please account for when each index was reported before treating the gap as
lag. One option is to use `lastHeartbeat`, which the active-store entry already
carries, to allow for the leader's progress between the two reports. Another is
to take follower progress from the leader itself, for example in its 5 s
partition heartbeat. Please also add a test where the target's report is older
than the leader's while the target is caught up.
--
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]
