arshilkxwork commented on code in PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#discussion_r4305939970
##########
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:
Fixed in e6c1bcb0d. `canTakeLeader` now skips a transfer when the target's
or the leader Store's report is more than 60 s old
(`BALANCE_LEADER_REPORT_MAX_AGE`, two Store heartbeats), before any backdating.
I applied it to the leader's report as well, since an old leader report says as
little about the leader's current index.
`testAutoBalancePartitionLeaderWithStaleReport` uses your case: after
`warmUp(service, 70_000, 300_000)`, leaders at 100000 and store 3 at 75000
reported 250 s ago, store 3 now waits and store 2 takes both transfers. Store 3
at 95000 reported 50 s ago still takes two leaders, and a leader report 90 s
old moves nothing.
##########
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:
Fixed in e6c1bcb0d along the lines of your first suggestion. Each group now
keeps the latest leader report a run saw and the one before it from the same
Store. When a run sees the same report as the previous run, it keeps both as
they are, so the rate still comes from an older sample.
`testAutoBalancePartitionLeaderWithShortInterval` runs three times about 10
s apart against reports with fixed times: 7000 at t-40 s, then 10000 at t-10 s
seen by the second and third runs. On the third run store 3 reports 8500 15 s
before the leader, is caught up at 100 entries/s, and takes two leaders. With
the old history that run had only the 10000 report, so it would compare raw
indexes 1500 apart and store 3 would wait.
##########
hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/core/TaskScheduleServiceTest.java:
##########
@@ -75,6 +87,466 @@ public void testBalancePartitionLeader() throws PDException
{
getStoreNodeService().getStoreInfoMeta().removeAll();
}
+ @Test
+ public void testAutoBalancePartitionLeaderWithinSkew() throws PDException {
Review Comment:
Added in e6c1bcb0d: `testAutoBalancePartitionLeaderWithSkewAtLimit` uses the
default `max-skew` of 2. Leaders at 3/1/2 (skew 2) return no transfers, and
4/1/1 (skew 3) plan transfers, none of them to store 1. With `<`, the skew-2
layout would reach the planner and get transfers, so the first assertion would
fail.
##########
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:
Done in e6c1bcb0d with your first option. The planner's target calculation
moved into a `leaderTargets` helper that the planner still uses, and the
periodic path skips the run, with one DEBUG line, when the leader counts
already equal those targets. The planner keeps its "shard group size" INFO line
but no longer logs the per-store shard counts there, so a skipped run logs
nothing at INFO.
`testAutoBalancePartitionLeaderWithLeadersOnTargets` sets `max-skew` to 0
with seven groups on three stores. Leaders already at the planner's 2/2/3
return no transfers. Leaders at 3/2/2 have the same spread on other stores, so
the planner runs and ends at 2/2/3.
##########
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:
Fixed in e6c1bcb0d: the row now says leaders move back "only through a
manual call: `/v1/task/balanceLeaders`, `/v1/balanceLeaders` or the
`balanceLeaders` gRPC call".
##########
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:
Fixed in e6c1bcb0d with your wording: "while a Store that holds a shard is
not in the active-store list, which it leaves `store.keepAlive-timeout` after
its last heartbeat". The same paragraph now also covers the 60 s report age and
the target-match skip.
--
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]
