bitflicker64 commented on code in PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#discussion_r4304095800
##########
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) ->
Review Comment:
Minor: The description says a layout whose best reachable spread is above
`max-skew` makes every run plan and move nothing. The cost is log volume: each
such run logs at INFO here, then the shard counts, the target counts and one
"is leader, no need change" line per group.
Example: 12 groups with 3 replicas on 5 Stores holding 8/7/7/7/7 shards. The
planner targets 3/3/2/2/2 leaders, so the skew stays 1, and with `max-skew: 0`,
which the docs allow, that repeats every interval.
Please skip the run when the leader counts already match the planner's
targets, or log the per-group line at DEBUG on the periodic path.
##########
hugegraph-pd/docs/configuration.md:
##########
@@ -188,6 +191,9 @@ pd:
| `pd.patrol-interval` | Integer | `1800` | Interval (in seconds) for
partition health patrol and automatic rebalancing. Lower values = more frequent
checks. |
| `pd.initial-store-count` | Integer | `1` | Minimum number of Store nodes
required for cluster to be operational. Set to expected initial store count. |
| `pd.initial-store-list` | String | `127.0.0.1:8500` | Comma-separated list
of Store gRPC addresses to auto-activate on startup. Useful for bootstrapping. |
+| `pd.balance-leader-enable` | Boolean | `false` | Run partition leader
balancing periodically on the PD leader. When off, leaders move back only
through `/v1/task/balanceLeaders`. |
Review Comment:
Minor: "leaders move back only through `/v1/task/balanceLeaders`" leaves out
the other two manual entries.
`StoreAPI` also serves `/v1/balanceLeaders`, and the `balanceLeaders` gRPC
call reaches `PDService.balanceLeaders`, which calls
`balancePartitionLeader(true)`. The PR description lists all three.
Please name all three, or say "only through a manual balanceLeaders call".
##########
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();
Review Comment:
Important: With `pd.balance-leader-interval` below the 30 s Store heartbeat,
two runs in a row can see the same leader report. This line then drops the
older sample, so the next run finds `previous.time() == leader.time()` and
compares raw indexes, as before 57bd3d19b.
I simulated ticks against 30 s heartbeats at 1000 entries/s with store 3
caught up, over 30 tick phases. With store 3 reporting 5 s before the leader,
store 3 got no leaders on any planning run in 15 of 30 phases at 10 s, and in
10 of 30 at 15 s and at 20 s. At 30 and 300 s it never was. In general, an
interval I below 30 s can defer a target that reports less than 30 - I s before
the leader.
This only slows balancing. Please keep the latest sample older than the
current report, or document a 30 s minimum for the interval.
--
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]
