Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[COST] limit brokerCost to ReplicaLeaderCost #1780

Open
wants to merge 7 commits into
base: main
Choose a base branch
from
Open

[COST] limit brokerCost to ReplicaLeaderCost #1780

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
0324123
limit brokerCost to ReplicaLeaderCost
qoo332001 May 26, 2023
79b41b9
update docs
qoo332001 May 29, 2023
e3dc0ef
update docs
qoo332001 May 29, 2023
46cb1e1
Merge branch 'main' of https://github.com/skiptests/astraea into revi…
qoo332001 May 29, 2023
3a38b03
fix conflict
qoo332001 May 29, 2023
55de744
fix bug
qoo332001 May 29, 2023
59c8be0
Merge branch 'main' of https://github.com/skiptests/astraea into revi…
qoo332001 May 30, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 22 additions & 1 deletion common/src/main/java/org/astraea/common/cost/ReplicaLeaderCost.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
package org.astraea.common.cost;

import static org.astraea.common.cost.MigrationCost.replicaLeaderToAdd;
import static org.astraea.common.cost.MigrationCost.replicaLeaderToRemove;

import java.util.List;
import java.util.Map;
Expand All @@ -32,13 +33,24 @@ public class ReplicaLeaderCost implements HasBrokerCost, HasClusterCost, HasMove
private final Dispersion dispersion = Dispersion.normalizedStandardDeviation();
private final Configuration config;
public static final String MAX_MIGRATE_LEADER_KEY = "max.migrated.leader.number";
static final String BROKER_COST_LIMIT_KEY = "max.broker.migrated.leader.number";
private final Map<Integer, Integer> brokerMoveCostLimit;

public ReplicaLeaderCost() {
this.config = new Configuration(Map.of());
this(Configuration.EMPTY);
}

public ReplicaLeaderCost(Configuration config) {
this.config = config;
this.brokerMoveCostLimit = brokerMoveCostLimit(config);
}

private Map<Integer, Integer> brokerMoveCostLimit(Configuration configuration) {
return configuration.list(BROKER_COST_LIMIT_KEY, ",").stream()
.collect(
Collectors.toMap(
idAndPath -> Integer.parseInt(idAndPath.split(":")[0]),
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

idAndPath? 這個命名是不是怪怪的

idAndPath -> Integer.parseInt(idAndPath.split(":")[1])));
}

@Override
Expand Down Expand Up @@ -79,8 +91,17 @@ public Configuration config() {
@Override
public MoveCost moveCost(ClusterInfo before, ClusterInfo after, ClusterBean clusterBean) {
var replicaLeaderIn = replicaLeaderToAdd(before, after);
var replicaLeaderOut = replicaLeaderToRemove(before, after);
var maxMigratedLeader =
config.string(MAX_MIGRATE_LEADER_KEY).map(Long::parseLong).orElse(Long.MAX_VALUE);
var brokerOverflow =
this.brokerMoveCostLimit.entrySet().stream()
.anyMatch(
leaderLimit ->
replicaLeaderIn.getOrDefault(leaderLimit.getKey(), 0L)
+ replicaLeaderOut.getOrDefault(leaderLimit.getKey(), 0L)
> leaderLimit.getValue());
if (brokerOverflow) return () -> true;
var overflow =
maxMigratedLeader
< replicaLeaderIn.values().stream().map(Math::abs).mapToLong(s -> s).sum();
Expand Down
27 changes: 26 additions & 1 deletion common/src/test/java/org/astraea/common/cost/ReplicaLeaderCostTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,8 @@ public class ReplicaLeaderCostTest {

@Test
void testLeaderCount() {

// test sum overflow
var baseCluster =
ClusterInfo.builder()
.addNode(Set.of(1, 2))
Expand Down Expand Up @@ -75,9 +77,32 @@ void testLeaderCount() {
new ReplicaLeaderCost(
new Configuration(Map.of(ReplicaLeaderCost.MAX_MIGRATE_LEADER_KEY, "10")))
.moveCost(sourceCluster, overFlowTargetCluster, ClusterBean.EMPTY);

Assertions.assertTrue(overFlowMoveCost.overflow());
Assertions.assertFalse(noOverFlowMoveCost.overflow());

// test broker overflow
var brokerTargetCluster =
ClusterInfo.builder(baseCluster)
.addTopic("topic1", 3, (short) 1, r -> Replica.builder(r).brokerId(1).build())
.addTopic("topic2", 3, (short) 1, r -> Replica.builder(r).brokerId(2).build())
.build();

var brokerOverFlowTargetCluster =
ClusterInfo.builder(baseCluster)
.addTopic("topic1", 3, (short) 1, r -> Replica.builder(r).brokerId(1).build())
.addTopic("topic2", 3, (short) 1, r -> Replica.builder(r).brokerId(1).build())
.build();
var overFlowMoveCost2 =
new ReplicaLeaderCost(
new Configuration(Map.of(ReplicaLeaderCost.BROKER_COST_LIMIT_KEY, "2:2")))
.moveCost(sourceCluster, brokerOverFlowTargetCluster, ClusterBean.EMPTY);

var noOverFlowMoveCost2 =
new ReplicaLeaderCost(
new Configuration(Map.of(ReplicaLeaderCost.BROKER_COST_LIMIT_KEY, "2:3")))
.moveCost(sourceCluster, brokerTargetCluster, ClusterBean.EMPTY);
Assertions.assertTrue(overFlowMoveCost2.overflow());
Assertions.assertFalse(noOverFlowMoveCost2.overflow());
}

@Test
Expand Down
17 changes: 9 additions & 8 deletions docs/web_server/web_api_balancer_chinese.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,14 +42,15 @@ POST /balancer

costConfig:

| config key | config value | value format |
| --------------------------- | ------------------------------------------------------------ | ------------------------------------------------------------ |
| max.migrated.size | 設定最大可搬移的資料量 | "`data size` + `unit`" ex.100KB, 500MB, 3GB |
| max.migrated.leader.number | 設定最大可搬移的leader 數量 | "`limit number`" ex. 1,2,3,100 |
| max.migrated.replica.number | 設定最大可搬移的replica 數量 | "`limit number`" ex. 1,2,3,100 |
| max.migrated.leader.size | 設定最大可搬移的leader 資料量 | "`data size` + `unit`" ex.100KB, 500MB, 3GB |
| max.broker.total.disk.space | 設定搬移過程中broker最大可以佔用的replica 資料量 | "`broker Id` + `:` + `data size` " ex. "0:1500MB ,1:1000MB ,2:1500MB" |
| max.broker.path.disk.space | 設定搬移過程中broker上的data folder最大可以佔用的replica 資料量 | "`broker Id` + `-` + `data path` + `:` + `data size` " ex. "0-/path0:1500MB,1-/path0:1000MB,2-/path0:1500MB,2-/path1:900MB" |
| config key | config value | value format |
| --------------------------------- | ------------------------------------------------------------ | ------------------------------------------------------------ |
| max.migrated.size | 設定最大可搬移的資料量 | "`data size` + `unit`" ex.100KB, 500MB, 3GB |
| max.migrated.leader.number | 設定最大可搬移的leader 數量 | "`limit number`" ex. 1,2,3,100 |
| max.broker.migrated.leader.number | 設定搬移過程中broker最大可搬移的leader數量 | "`broker Id` + `:` + `limit number`" ex. "0:150 ,1:30" |
| max.migrated.replica.number | 設定最大可搬移的replica 數量 | "`limit number`" ex. 1,2,3,100 |
| max.migrated.leader.size | 設定最大可搬移的leader 資料量 | "`data size` + `unit`" ex.100KB, 500MB, 3GB |
| max.broker.total.disk.space | 設定搬移過程中broker最大可以佔用的replica 資料量 | "`broker Id` + `:` + `data size` " ex. "0:1500MB ,1:1000MB ,2:1500MB" |
| max.broker.path.disk.space | 設定搬移過程中broker上的data folder最大可以佔用的replica 資料量 | "`broker Id` + `-` + `data path` + `:` + `data size` " ex. "0-/path0:1500MB,1-/path0:1000MB,2-/path0:1500MB,2-/path1:900MB" |

目前支援的 Cost Function

Expand Down