arshilkxwork commented on code in PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#discussion_r4300463681
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java:
##########
@@ -582,6 +607,80 @@ 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.
+ *
+ * @return shard group id -> the store id its leader was moved to; empty
when skipped
+ */
+ public Map<Integer, Long> autoBalancePartitionLeader() throws PDException {
+ if (!isLeader()) {
+ 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 <= pdConfig.getBalanceLeaderMaxSkew()) {
+ 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, pdConfig.getBalanceLeaderMaxSkew());
+ return balancePartitionLeader(false);
Review Comment:
Done in 3daaa200f. `autoBalancePartitionLeader()` now hands the groups it
checked to the planner through a private `balancePartitionLeader` overload; the
public method still reads them itself, so the manual path is unchanged. It is
`synchronized` too, like the manual entry, so a manual call can't run between
the check and the plan. Right before each transfer, `canTakeLeader` reads the
group again and sends nothing if any field differs from the checked copy,
including shard roles, `conf_ver` and a leader that moved in the meantime.
`testAutoBalancePartitionLeaderWithChangeAfterCheck` turns store 3's shards
into `None` members right after the balance reads the group list and expects no
transfers.
That still leaves the time between the reread and the Store handling the
instruction. JRaft 1.3.14 covers it on the Store side:
`NodeImpl.transferLeadershipTo` refuses while the leader is changing its
configuration and refuses a peer outside the current configuration, so a `None`
or Learner shard can't receive leadership.
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java:
##########
@@ -582,6 +607,80 @@ 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.
+ *
+ * @return shard group id -> the store id its leader was moved to; empty
when skipped
+ */
+ public Map<Integer, Long> autoBalancePartitionLeader() throws PDException {
+ if (!isLeader()) {
+ 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 &&
Review Comment:
Fixed for the periodic path in 3daaa200f. A planned transfer now also needs
the target's applied index to be within 1000 entries of the leader's. The
indexes come from the `RaftStats` each Store sends with its store heartbeat
(`PartitionStateMachine.committedIndex`, set in `onApply`), read through
`getCommittedIndexStats()`. A target that reports no index is skipped, and so
is a group without a leader shard. A skipped transfer still uses its slot in
the plan, so that leader stays put until a later run and the other groups are
balanced as usual.
The indexes can be up to 30 s old, since Store heartbeats run every 30
seconds, and the two reports aren't taken at the same moment. When the leader's
report is newer, a caught-up follower can look behind, and the transfer is only
deferred. When the follower's report is newer, a follower that fell behind
after reporting can still pass. JRaft then times the transfer out after
`electionTimeoutMs`, and the old leader returns to `STATE_LEADER` and accepts
writes again. Closing that gap would need the Store to check replicator
progress before accepting a transfer; I'd leave that out of this PR. The manual
`balanceLeaders` path is unchanged.
`testAutoBalancePartitionLeaderWithLaggingFollower` puts all six leaders on
store 1. With store 3 5000 entries behind, and again with store 3 reporting no
index, there are two transfers, both to store 2. With store 3 500 behind there
are four, two of them to store 3.
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/config/PDConfig.java:
##########
@@ -58,6 +58,16 @@ public class PDConfig implements InitializingBean {
// The patrol task interval
@Value("${pd.patrol-interval:300}")
private long patrolInterval = 300;
+
+ // Periodic partition leader balancing, off by default; the interval is in
seconds
+ @Value("${pd.balance-leader-enable:false}")
+ private boolean balanceLeaderEnable = false;
+ @Value("${pd.balance-leader-interval:300}")
+ private long balanceLeaderInterval = 300;
+ // Skip the periodic run while max - min leaders per store is at or below
this
+ @Value("${pd.balance-leader-max-skew:2}")
+ private int balanceLeaderMaxSkew = 2;
Review Comment:
In 3daaa200f, `init()` schedules the periodic balance only when
`pd.balance-leader-interval` is positive and `pd.balance-leader-max-skew` is 0
or more. Otherwise it logs one warning with both values and schedules nothing.
The PR already handled a non-positive interval this way: a bad value turns the
feature off and PD still starts. `autoBalancePartitionLeader()` also returns
without planning on a negative skew, in case something calls it directly.
`configuration.md` now gives the allowed range for both keys.
`testAutoBalancePartitionLeaderWithNegativeSkew` sets -1 on a layout that moves
four leaders at the default skew and expects no transfers.
--
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]
