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

New cluster cost based on replica number #811

Merged
merged 1 commit into from
Oct 2, 2022
Merged

New cluster cost based on replica number #811

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
21 changes: 20 additions & 1 deletion ...g/astraea/common/cost/ReplicaNumCost.java → ...straea/common/cost/ReplicaNumberCost.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,16 +17,18 @@
package org.astraea.common.cost;

import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.astraea.common.admin.ClusterBean;
import org.astraea.common.admin.ClusterInfo;
import org.astraea.common.admin.Replica;
import org.astraea.common.metrics.collector.Fetcher;

/** more replicas migrate -> higher cost */
public class ReplicaNumCost implements HasMoveCost.Helper {
public class ReplicaNumberCost implements HasClusterCost, HasMoveCost.Helper {

@Override
public Optional<Fetcher> fetcher() {
Expand All @@ -50,4 +52,21 @@ public MoveCost moveCost(
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue, Long::sum)))
.build();
}

@Override
public ClusterCost clusterCost(ClusterInfo<Replica> clusterInfo, ClusterBean clusterBean) {
var group =
clusterInfo.replicas().stream().collect(Collectors.groupingBy(r -> r.nodeInfo().id()));

// there is no better plan for single node
if (clusterInfo.nodes().size() == 1) return () -> 0;

// worst case: all partitions are hosted by single node
if (clusterInfo.nodes().size() > 1 && group.size() <= 1) return () -> Long.MAX_VALUE;

// normal case
var max = group.values().stream().mapToLong(List::size).max().orElse(0);
var min = group.values().stream().mapToLong(List::size).min().orElse(0);
return () -> max - min;
}
}
43 changes: 39 additions & 4 deletions .../astraea/app/cost/ReplicaNumCostTest.java → ...ea/common/cost/ReplicaNumberCostTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,21 +14,56 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.astraea.app.cost;
package org.astraea.common.cost;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import org.astraea.common.admin.ClusterBean;
import org.astraea.common.admin.ClusterInfo;
import org.astraea.common.admin.NodeInfo;
import org.astraea.common.admin.Replica;
import org.astraea.common.cost.ReplicaNumCost;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;

class ReplicaNumCostTest {
class ReplicaNumberCostTest {

@Test
void testClusterCostForSingleNode() {
var nodeInfo = NodeInfo.of(10, "h", 100);
var clusterInfo = ClusterInfo.of(Set.of(nodeInfo), List.<Replica>of());
var cost = new ReplicaNumberCost();
Assertions.assertEquals(0, cost.clusterCost(clusterInfo, ClusterBean.EMPTY).value());
}

@Test
void testClusterCostForAllInSingleNode() {
var nodeInfo = NodeInfo.of(10, "h", 100);
var replica = Replica.builder().nodeInfo(nodeInfo).topic("t").build();
var clusterInfo = ClusterInfo.of(Set.of(nodeInfo, NodeInfo.of(11, "j", 100)), List.of(replica));
var cost = new ReplicaNumberCost();
Assertions.assertEquals(
Long.MAX_VALUE, cost.clusterCost(clusterInfo, ClusterBean.EMPTY).value());
}

@Test
void testClusterCost() {
var nodeInfo = NodeInfo.of(10, "h", 100);
var replicas =
IntStream.range(0, 10)
.mapToObj(i -> Replica.builder().nodeInfo(nodeInfo).topic("t").build())
.collect(Collectors.toCollection(ArrayList::new));
replicas.add(Replica.builder().nodeInfo(NodeInfo.of(11, "j", 100)).topic("t").build());
var clusterInfo = ClusterInfo.of(Set.of(nodeInfo, NodeInfo.of(11, "j", 100)), replicas);
var cost = new ReplicaNumberCost();
Assertions.assertEquals(9, cost.clusterCost(clusterInfo, ClusterBean.EMPTY).value());
}

@Test
void testMoveCost() {
var costFunction = new ReplicaNumCost();
var costFunction = new ReplicaNumberCost();
var before =
List.of(
Replica.of(
Expand Down