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

[Feature]集群均衡支持LogDir级别的Disk均衡(#1164) #1166

Closed
wants to merge 1 commit into from
Closed

[Feature]集群均衡支持LogDir级别的Disk均衡(#1164) #1166

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
10 changes: 10 additions & 0 deletions ...ava/com/xiaojukeji/know/streaming/km/rebalance/algorithm/executor/common/BalanceTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ public class BalanceTask {
private int partition;
//副本分配列表
private List<Integer> replicas;
private List<String> logDirs;

public String getTopic() {
return topic;
Expand All @@ -32,12 +33,21 @@ public void setReplicas(List<Integer> replicas) {
this.replicas = replicas;
}

public List<String> getLogDirs() {
return logDirs;
}

public void setLogDirs(List<String> logDirs) {
this.logDirs = logDirs;
}

@Override
public String toString() {
return "BalanceTask{" +
"topic='" + topic + '\'' +
", partition=" + partition +
", replicas=" + replicas +
", logDirs=" + logDirs +
'}';
}
}
2 changes: 2 additions & 0 deletions ...com/xiaojukeji/know/streaming/km/rebalance/algorithm/executor/common/OptimizerResult.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -120,6 +120,8 @@ public List<BalanceTask> resultTask() {
task.setPartition(proposal.tp().partition());
List<Integer> replicas = proposal.newReplicas().stream().map(ReplicaPlacementInfo::brokerId).collect(Collectors.toList());
task.setReplicas(replicas);
List<String> logDirs = proposal.newReplicas().stream().map(ReplicaPlacementInfo::logdir).collect(Collectors.toList());
task.setLogDirs(logDirs);
balanceTasks.add(task);
});
return balanceTasks;
Expand Down
37 changes: 30 additions & 7 deletions ...ance/src/main/java/com/xiaojukeji/know/streaming/km/rebalance/algorithm/model/Broker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import org.apache.kafka.common.TopicPartition;

import java.util.*;
import java.util.concurrent.ThreadLocalRandom;
import java.util.function.Predicate;
import java.util.stream.Collectors;

Expand All @@ -11,19 +12,16 @@
* @date 2022/4/29
*/
public class Broker implements Comparable<Broker> {
public static final Broker NONE = new Broker(new Rack("-1"), -1, "localhost", true, new Capacity());

private final Rack rack;
private final int id;
private final String host;
private final boolean isOffline;

private final Set<Replica> replicas;
private final Set<Replica> leaderReplicas;
private final Map<String, Map<Integer, Replica>> topicReplicas;

private final Map<String, LogDir> logDirs;
private final Load load;

private final Capacity capacity;

public Broker(Rack rack, int id, String host, boolean isOffline, Capacity capacity) {
Expand All @@ -34,10 +32,15 @@ public Broker(Rack rack, int id, String host, boolean isOffline, Capacity capaci
this.replicas = new HashSet<>();
this.leaderReplicas = new HashSet<>();
this.topicReplicas = new HashMap<>();
this.logDirs = new HashMap<>();
this.load = new Load();
this.capacity = capacity;
}

public void addLogDir(LogDir logDir) {
this.logDirs.put(logDir.name(), logDir);
}

public Rack rack() {
return rack;
}
Expand All @@ -58,6 +61,10 @@ public Set<Replica> replicas() {
return Collections.unmodifiableSet(this.replicas);
}

public Map<String, LogDir> logDirs() {
return Collections.unmodifiableMap(this.logDirs);
}

public SortedSet<Replica> sortedReplicasFor(Resource resource, boolean reverse) {
return sortedReplicasFor(null, resource, reverse);
}
Expand All @@ -84,27 +91,34 @@ public Set<Replica> leaderReplicas() {
}

public Load load() {
return load;
return this.load;
}

public Capacity capacity() {
return capacity;
}


public double utilizationFor(Resource resource) {
return this.load.loadFor(resource) / this.capacity.capacityFor(resource);
}


public double expectedUtilizationAfterAdd(Resource resource, Load loadToChange) {
return (this.load.loadFor(resource) + ((loadToChange == null) ? 0 : loadToChange.loadFor(resource)))
/ this.capacity.capacityFor(resource);
}


public double expectedUtilizationAfterRemove(Resource resource, Load loadToChange) {
return (this.load.loadFor(resource) - ((loadToChange == null) ? 0 : loadToChange.loadFor(resource)))
/ this.capacity.capacityFor(resource);
}

public LogDir logDir(String name) {
return this.logDirs.get(name);
}

public Replica replica(TopicPartition topicPartition) {
Map<Integer, Replica> replicas = this.topicReplicas.get(topicPartition.topic());
if (replicas == null) {
Expand All @@ -113,7 +127,7 @@ public Replica replica(TopicPartition topicPartition) {
return replicas.get(topicPartition.partition());
}

void addReplica(Replica replica) {
void addReplica(String logDir, Replica replica) {
// Add replica to list of all replicas in the broker.
if (this.replicas.contains(replica)) {
throw new IllegalStateException(String.format("Broker %d already has replica %s", this.id,
Expand All @@ -131,9 +145,12 @@ void addReplica(Replica replica) {

// Add replica load to the broker load.
this.load.addLoad(replica.load());

// Add replica to list of replicas in the logDir
this.logDirs.get(logDir).addReplica(replica);
}

Replica removeReplica(TopicPartition topicPartition) {
Replica removeReplica(String sourceLogDir, TopicPartition topicPartition) {
Replica replica = replica(topicPartition);
if (replica != null) {
this.replicas.remove(replica);
Expand All @@ -145,6 +162,7 @@ Replica removeReplica(TopicPartition topicPartition) {
this.leaderReplicas.remove(replica);
}
this.load.subtractLoad(replica.load());
this.logDirs.get(sourceLogDir).removeReplica(topicPartition);
}
return replica;
}
Expand Down Expand Up @@ -193,6 +211,7 @@ public String toString() {
", replicas=" + replicas +
", leaderReplicas=" + leaderReplicas +
", topicReplicas=" + topicReplicas +
", logDirs=" + logDirs +
", load=" + load +
", capacity=" + capacity +
'}';
Expand All @@ -219,4 +238,8 @@ public Collection<Replica> replicasOfTopicInBroker(String topic) {
public Set<Replica> currentOfflineReplicas() {
return replicas.stream().filter(Replica::isCurrentOffline).collect(Collectors.toSet());
}

public LogDir randomLogDir() {
return logDirs.values().stream().collect(Collectors.toList()).get(ThreadLocalRandom.current().nextInt(logDirs.size()));
}
}
53 changes: 38 additions & 15 deletions ...rc/main/java/com/xiaojukeji/know/streaming/km/rebalance/algorithm/model/ClusterModel.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,25 @@ public SortedSet<Broker> sortedBrokersFor(Predicate<? super Broker> filter, Reso
return sortedBrokers;
}

public SortedSet<LogDir> sortedLogDirsFor(Predicate<? super LogDir> filter, Resource resource, boolean reverse) {
Comparator<LogDir> comparator =
Comparator.<LogDir>comparingDouble(b -> b.utilizationFor(resource));
if (reverse)
comparator = comparator.reversed();

SortedSet<LogDir> sortedLogDirs = new TreeSet<>(comparator);
this.brokersById.values().stream().forEach(broker -> {
if (filter == null) {
sortedLogDirs.addAll(broker.logDirs().values());
} else {
sortedLogDirs.addAll(broker.logDirs().values().stream()
.filter(filter).collect(Collectors.toList()));
}
});

return sortedLogDirs;
}

public Load load() {
Load load = new Load();
for (Broker broker : this.brokersById.values()) {
Expand Down Expand Up @@ -101,30 +120,33 @@ public Broker broker(int brokerId) {
return this.brokersById.get(brokerId);
}

public Broker addBroker(String rackId, int brokerId, String host, boolean isOffline, Capacity capacity) {
public Broker addBroker(String rackId, int brokerId, String host, boolean isOffline, Capacity capacity, Map<String, Capacity> logDirCapacities) {
Rack rack = rack(rackId);
if (rack == null)
throw new IllegalArgumentException("Rack: " + rackId + "is not exists.");
Broker broker = new Broker(rack, brokerId, host, isOffline, capacity);
for(Map.Entry<String, Capacity> entry : logDirCapacities.entrySet()) {
broker.addLogDir(new LogDir(entry.getKey(), broker, entry.getValue()));
}
rack.addBroker(broker);
this.brokersById.put(brokerId, broker);
return broker;
}

public Replica addReplica(int brokerId, TopicPartition topicPartition, boolean isLeader, Load load) {
return addReplica(brokerId, topicPartition, isLeader, false, load);
public Replica addReplica(int brokerId, String logDir, TopicPartition topicPartition, boolean isLeader, Load load) {
return addReplica(brokerId, logDir, topicPartition, isLeader, false, load);
}

public Replica addReplica(int brokerId, TopicPartition topicPartition, boolean isLeader, boolean isOffline, Load load) {
public Replica addReplica(int brokerId, String logDir, TopicPartition topicPartition, boolean isLeader, boolean isOffline, Load load) {
Broker broker = broker(brokerId);
if (broker == null) {
throw new IllegalArgumentException("Broker: " + brokerId + "is not exists.");
}

Replica replica = new Replica(broker, topicPartition, isLeader, isOffline);
Replica replica = new Replica(broker, logDir, topicPartition, isLeader, isOffline);
replica.setLoad(load);
// add to broker
broker.addReplica(replica);
broker.addReplica(logDir, replica);

Map<TopicPartition, Partition> partitions = this.partitionsByTopic
.computeIfAbsent(topicPartition.topic(), k -> new HashMap<>());
Expand All @@ -139,9 +161,9 @@ public Replica addReplica(int brokerId, TopicPartition topicPartition, boolean i
return replica;
}

public Replica removeReplica(int brokerId, TopicPartition topicPartition) {
public Replica removeReplica(int brokerId, String sourceLogDir, TopicPartition topicPartition) {
Broker broker = broker(brokerId);
return broker.removeReplica(topicPartition);
return broker.removeReplica(sourceLogDir, topicPartition);
}

public void relocateLeadership(String goal, String actionType, TopicPartition topicPartition, int sourceBrokerId, int destinationBrokerId) {
Expand Down Expand Up @@ -171,19 +193,20 @@ public void relocateLeadership(TopicPartition topicPartition, int sourceBrokerId
partition.relocateLeadership(destinationReplica);
}

public void relocateReplica(String goal, String actionType, TopicPartition topicPartition, int sourceBrokerId, int destinationBrokerId) {
relocateReplica(topicPartition, sourceBrokerId, destinationBrokerId);
public void relocateReplica(String goal, String actionType, TopicPartition topicPartition, int sourceBrokerId, String sourceLogDir, int destinationBrokerId, String destinationLogDir) {
relocateReplica(topicPartition, sourceBrokerId, sourceLogDir, destinationBrokerId, destinationLogDir);
addBalanceActionHistory(goal, actionType, topicPartition, sourceBrokerId, destinationBrokerId);
}

public void relocateReplica(TopicPartition topicPartition, int sourceBrokerId, int destinationBrokerId) {
Replica replica = removeReplica(sourceBrokerId, topicPartition);
public void relocateReplica(TopicPartition topicPartition, int sourceBrokerId, String sourceLogDir, int destinationBrokerId, String destinationLogDir) {
Replica replica = removeReplica(sourceBrokerId, sourceLogDir, topicPartition);
if (replica == null) {
throw new IllegalArgumentException("Replica is not in the cluster.");
}
Broker destinationBroker = broker(destinationBrokerId);
replica.setBroker(destinationBroker);
destinationBroker.addReplica(replica);
replica.setLogDir(destinationLogDir);
destinationBroker.addReplica(destinationLogDir, replica);
}

private void addBalanceActionHistory(String goal, String actionType, TopicPartition topicPartition, int sourceBrokerId, int destinationBrokerId) {
Expand All @@ -208,7 +231,7 @@ public Map<TopicPartition, List<ReplicaPlacementInfo>> getReplicaDistribution()
for (Map<TopicPartition, Partition> tp : partitionsByTopic.values()) {
tp.values().forEach(i -> {
i.replicas().forEach(j -> replicaDistribution.computeIfAbsent(j.topicPartition(), k -> new ArrayList<>())
.add(new ReplicaPlacementInfo(j.broker().id(), "")));
.add(new ReplicaPlacementInfo(j.broker().id(), j.logDir())));
});
}
return replicaDistribution;
Expand All @@ -221,7 +244,7 @@ public Replica partition(TopicPartition tp) {
public Map<TopicPartition, ReplicaPlacementInfo> getLeaderDistribution() {
Map<TopicPartition, ReplicaPlacementInfo> leaderDistribution = new HashMap<>();
for (Broker broker : brokersById.values()) {
broker.leaderReplicas().forEach(i -> leaderDistribution.put(i.topicPartition(), new ReplicaPlacementInfo(broker.id(), "")));
broker.leaderReplicas().forEach(i -> leaderDistribution.put(i.topicPartition(), new ReplicaPlacementInfo(broker.id(), i.logDir())));
}
return leaderDistribution;
}
Expand Down
Loading
Loading