arshilkxwork commented on code in PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#discussion_r4303294367
##########
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) {
Review Comment:
Fixed in 57bd3d19b. `canTakeLeader` now checks `isLeader()` first, right
before each periodic transfer goes to `PartitionService.transferLeader()`, so a
PD that lost leadership during the run sends nothing it planned.
`testAutoBalancePartitionLeaderAfterLosingLeadership` uses a service whose
`isLeader()` is true only for the entry check: the run sends no transfers, and
the test asserts the planned transfers reached the recheck. `isLeader()` is
`protected` now so the test can override it.
##########
hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/core/TaskScheduleServiceTest.java:
##########
@@ -75,6 +83,217 @@ public void testBalancePartitionLeader() throws PDException
{
getStoreNodeService().getStoreInfoMeta().removeAll();
}
+ @Test
+ public void testAutoBalancePartitionLeaderWithinSkew() throws PDException {
+ // leaders spread 2/2/2 over the three stores
+ var rst = autoBalance(new long[]{1, 2, 3, 1, 2, 3}, 1, 2, 3);
+ assertTrue(rst.isEmpty());
+ }
+
+ @Test
+ public void testAutoBalancePartitionLeaderOverSkew() throws PDException {
+ // every leader on store 1, as after stores 2 and 3 restarted
+ var rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1}, 1, 2, 3);
+ assertEquals(4, rst.size());
+ assertFalse(rst.containsValue(1L));
+ }
+
+ @Test
+ public void testAutoBalancePartitionLeaderWithInactiveStore() throws
PDException {
+ // store 3 holds shards but sends no heartbeat, so it cannot take a
leader
+ var rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1}, 1, 2);
+ assertTrue(rst.isEmpty());
+ }
+
+ @Test
+ public void testAutoBalancePartitionLeaderWithWarnGroup() throws
PDException {
+ // store 3 just stopped: still in the active set, but its groups are
already Warn
+ var rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1},
+ g ->
g.toBuilder().setState(Metapb.PartitionState.PState_Warn)
+ .build(), 0, 1, 2, 3);
+ assertTrue(rst.isEmpty());
+
+ // once the group is Normal again, the same imbalance is balanced
+ rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1}, g -> g, 0, 1, 2, 3);
+ assertEquals(4, rst.size());
+ }
+
+ @Test
+ public void testAutoBalancePartitionLeaderWithoutQuorum() throws
PDException {
+ // group 0 lost its quorum: no partition heartbeat for a minute, while
its last state
+ // is still Normal and every store is still in the active set
+ var rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1}, g -> g, 60 * 1000,
1, 2, 3);
+ assertTrue(rst.isEmpty());
+
+ // group 0 never reported stats at all
+ rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1}, g -> g, -1, 1, 2, 3);
+ assertTrue(rst.isEmpty());
+
+ // a heartbeat within the window: the same imbalance is balanced
+ rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1}, g -> g, 10 * 1000, 1,
2, 3);
+ assertEquals(4, rst.size());
+ }
+
+ @Test
+ public void testAutoBalancePartitionLeaderWithPendingMember() throws
PDException {
+ // reallocShards added the shard on store 3 and left its role None
+ for (var role : List.of(Metapb.ShardRole.None,
Metapb.ShardRole.Learner)) {
+ var rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1},
+ g -> g.toBuilder().setShards(2,
g.getShards(2).toBuilder()
+
.setRole(role))
+ .build(), 0, 1, 2, 3);
+ assertTrue(role.name(), rst.isEmpty());
+ }
+ }
+
+ @Test
+ public void testAutoBalancePartitionLeaderWithChangeAfterCheck() throws
PDException {
+ // reallocShards turns store 3's shards into new None members right
after the check
+ // read the groups: the plan is made from the checked groups and no
transfer is sent
+ var rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1}, g -> g, 0,
+ (storeId, groupId) -> 10_000,
+ g -> g.toBuilder().setShards(2,
g.getShards(2).toBuilder()
+
.setRole(Metapb.ShardRole.None))
+ .build(), 1, 2, 3);
+ assertTrue(rst.isEmpty());
+ }
+
+ @Test
+ public void testAutoBalancePartitionLeaderWithLaggingFollower() throws
PDException {
+ // store 3's replicas trail the leaders by 5000 entries or report no
index: its share
+ // of the leaders stays on store 1 and only store 2 takes leaders
+ for (long store3Index : new long[]{5_000, -1}) {
+ var rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1}, g -> g, 0,
+ (storeId, groupId) -> storeId == 3 ?
store3Index : 10_000,
+ null, 1, 2, 3);
+ assertEquals(2, rst.size());
+ assertTrue(rst.values().stream().allMatch(storeId -> storeId ==
2L));
+ }
+
+ // 500 entries behind is close enough to take a leader
+ var rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1}, g -> g, 0,
+ (storeId, groupId) -> storeId == 3 ? 9_500 :
10_000,
+ null, 1, 2, 3);
+ assertEquals(4, rst.size());
+ assertTrue(rst.containsValue(3L));
+ }
+
+ @Test
+ public void testAutoBalancePartitionLeaderWithNegativeSkew() throws
PDException {
+ // with the default skew this layout moves four leaders; -1 must not
run a plan
+ getPdConfig().setBalanceLeaderMaxSkew(-1);
+ try {
+ var rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1}, 1, 2, 3);
+ assertTrue(rst.isEmpty());
+ } finally {
+ getPdConfig().setBalanceLeaderMaxSkew(2);
+ }
+ }
+
+ private Map<Integer, Long> autoBalance(long[] leaderStores, long...
activeStores) throws
+
PDException {
+ return autoBalance(leaderStores, g -> g, 0, activeStores);
+ }
+
+ private Map<Integer, Long> autoBalance(long[] leaderStores,
+ UnaryOperator<Metapb.ShardGroup>
group0,
+ long group0StatsAge,
+ long... activeStores) throws
PDException {
+ return autoBalance(leaderStores, group0, group0StatsAge, (storeId,
groupId) -> 10_000,
+ null, activeStores);
+ }
+
+ /**
+ * Stores shard groups 0..n-1 on stores 1, 2 and 3 with the leader of
group i on
+ * leaderStores[i], all Normal with fresh partition stats, then applies
group0 to group 0
+ * and ages its stats by group0StatsAge ms (-1: no stats). Marks
activeStores alive, each
+ * reporting appliedIndex(store, group) for every group (negative: not
reported), and runs
+ * one periodic balance. A non-null afterRead is applied to every stored
group right after
+ * the balance reads the group list, like a change landing between check
and transfer.
+ */
+ private Map<Integer, Long> autoBalance(long[] leaderStores,
+ UnaryOperator<Metapb.ShardGroup>
group0,
+ long group0StatsAge,
+ ToLongBiFunction<Long, Integer>
appliedIndex,
+ UnaryOperator<Metapb.ShardGroup>
afterRead,
+ long... activeStores) throws
PDException {
+ var storeMeta = getStoreNodeService().getStoreInfoMeta();
+ var partitionMeta = MetadataFactory.newPartitionMeta(getPdConfig());
+ // Set aside groups other tests in the suite left behind, so only ours
are planned
+ var otherGroups = storeMeta.getShardGroups();
+ for (var group : otherGroups) {
+ storeMeta.deleteShardGroup(group.getId());
+ }
+ var partitions = new ArrayList<Metapb.Partition>();
+ long now = System.currentTimeMillis();
+ for (int i = 0; i < leaderStores.length; i++) {
+ var group = genShardGroup(i, leaderStores[i],
Metapb.PartitionState.PState_Normal);
+ long statsAge = 0;
+ if (i == 0) {
+ group = group0.apply(group);
+ statsAge = group0StatsAge;
+ }
+ storeMeta.updateShardGroup(group);
+ if (statsAge >= 0) {
+
partitionMeta.updatePartitionStats(Metapb.PartitionStats.newBuilder()
+
.setId(i)
+
.setTimestamp(
+
now - statsAge)
+
.build());
+ }
+ partitions.add(genPartition(i));
+ }
+ for (long storeId = 1; storeId <= 3; storeId++) {
+ var stats = Metapb.StoreStats.newBuilder();
+ for (int i = 0; i < leaderStores.length; i++) {
+ long index = appliedIndex.applyAsLong(storeId, i);
+ if (index >= 0) {
+
stats.addRaftStats(Metapb.RaftStats.newBuilder().setPartitionId(i)
+
.setCommittedIndex(index));
+ }
+ }
+ var store =
Metapb.Store.newBuilder().setId(storeId).setStats(stats).build();
+ if (Arrays.stream(activeStores).anyMatch(id -> id ==
store.getId())) {
+ storeMeta.keepStoreAlive(store);
+ } else {
+ storeMeta.removeActiveStore(store);
+ }
+ }
+ getPdConfig().getPartition().setShardCount(3);
+ getPartitionService().updatePartition(partitions);
+ try {
+ StoreNodeService storeService = getStoreNodeService();
+ if (afterRead != null) {
+ storeService = new StoreNodeService(getPdConfig()) {
+ @Override
+ public List<Metapb.ShardGroup> getShardGroups() throws
PDException {
+ var groups = super.getShardGroups();
+ for (var group : groups) {
+ storeMeta.updateShardGroup(afterRead.apply(group));
+ }
+ return groups;
+ }
+ };
+ }
+ // A fresh service, so a balance run by another test does not hit
the rate limit
+ var service = new TaskScheduleService(getPdConfig(), storeService,
+ getPartitionService());
+ return service.autoBalancePartitionLeader();
Review Comment:
Added in 57bd3d19b: `testInitSchedulesLeaderBalance` calls `init()` on a
service whose `autoBalancePartitionLeader()` only counts down a latch.
Disabled, interval 0, interval -5 and skew -1 leave the timer unscheduled. Skew
0 and skew 2 with a 1 s interval schedule it, and the latch must fire within 10
s. `init()` keeps the scheduled future and `isLeaderBalanceScheduled()` reports
whether there is one, which is what the test checks for each case.
##########
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:
Done in 57bd3d19b, along the lines of your first option. The check now reads
each Store's `lastHeartbeat` along with its `RaftStats`, and every periodic run
keeps the leader Store's report for each group, including runs that skip. On
the next run the leader's write rate is (index now - index then) / (time now -
time then), and PD moves the leader's index to the time of the target's report
at that rate before comparing. A target whose report is d seconds older is
compared with the leader's estimated index d seconds earlier, so the fixed
heartbeat phase no longer shows up as lag on every run. Until a run has seen a
rate (the first run after this PD becomes leader, or after the group's leader
moved), the indexes are compared as reported, so a caught-up Store can wait one
extra run. The rate is an average over the balance interval, so bursty writes
still add some error; the 1000-entry gap leaves room for part of it.
`testAutoBalancePartitionLeaderWithOlderFollowerReport`: the leaders on
store 1 apply 100 entries/s and store 3's heartbeat is 15 s older, so its
caught-up replicas report 8500 against 10000. With no earlier run, store 3
waits: two transfers, both to store 2. After a run 30 s earlier that saw the
leaders at 7000, the same reports give four transfers, two of them to store 3.
A report of 5000 from store 3 at that rate is 3500 behind and still waits.
--
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]
