Add endpoint to move all replicas from a disk to another disk of the same broker #7
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,140 @@ | ||
| /* | ||
| * Copyright 2018 LinkedIn Corp. Licensed under the BSD 2-Clause License (the "License"). See License in the project root for license information. | ||
| */ | ||
|
|
||
| package com.linkedin.kafka.cruisecontrol.analyzer.goals; | ||
|
|
||
| import com.linkedin.kafka.cruisecontrol.analyzer.OptimizationOptions; | ||
| import com.linkedin.kafka.cruisecontrol.analyzer.ActionAcceptance; | ||
| import com.linkedin.kafka.cruisecontrol.analyzer.BalancingAction; | ||
| import com.linkedin.kafka.cruisecontrol.analyzer.ProvisionResponse; | ||
| import com.linkedin.kafka.cruisecontrol.analyzer.ProvisionStatus; | ||
| import com.linkedin.kafka.cruisecontrol.model.Broker; | ||
| import com.linkedin.kafka.cruisecontrol.model.ClusterModel; | ||
| import com.linkedin.kafka.cruisecontrol.model.ClusterModelStats; | ||
| import com.linkedin.kafka.cruisecontrol.model.Disk; | ||
| import com.linkedin.kafka.cruisecontrol.model.Replica; | ||
| import com.linkedin.kafka.cruisecontrol.monitor.ModelCompletenessRequirements; | ||
| import java.util.Set; | ||
| import java.util.Map; | ||
| import java.util.Collections; | ||
| import java.util.List; | ||
| import java.util.ArrayList; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import static com.linkedin.kafka.cruisecontrol.analyzer.ActionAcceptance.ACCEPT; | ||
| import static com.linkedin.kafka.cruisecontrol.analyzer.goals.GoalUtils.MIN_NUM_VALID_WINDOWS_FOR_SELF_HEALING; | ||
|
|
||
|
|
||
| /** | ||
| * Soft goal to move the replicas to different log dir. | ||
| */ | ||
| public class DiskRemovalGoal implements Goal { | ||
| private static final Logger LOG = LoggerFactory.getLogger(DiskRemovalGoal.class); | ||
| private final ProvisionResponse _provisionResponse; | ||
|
|
||
| protected final Map<Integer, Set<String>> _brokerIdAndLogdirs; | ||
|
|
||
| public DiskRemovalGoal() { | ||
| this(Collections.emptyMap()); | ||
| } | ||
|
|
||
| public DiskRemovalGoal(Map<Integer, Set<String>> brokerIdAndLogdirs) { | ||
| _provisionResponse = new ProvisionResponse(ProvisionStatus.UNDECIDED); | ||
| _brokerIdAndLogdirs = brokerIdAndLogdirs; | ||
| } | ||
|
|
||
| private void sanityCheckOptimizationOptions(OptimizationOptions optimizationOptions) { | ||
| if (optimizationOptions.isTriggeredByGoalViolation()) { | ||
| throw new IllegalArgumentException(String.format("%s goal does not support use by goal violation detector.", name())); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public boolean optimize(ClusterModel clusterModel, Set<Goal> optimizedGoals, OptimizationOptions optimizationOptions) { | ||
| sanityCheckOptimizationOptions(optimizationOptions); | ||
|
|
||
| for (Map.Entry<Integer, Set<String>> brokerIdLogDirs : _brokerIdAndLogdirs.entrySet()) { | ||
| Integer brokerId = brokerIdLogDirs.getKey(); | ||
| Set<String> logDirs = brokerIdLogDirs.getValue(); | ||
|
|
||
| Broker currentBroker = clusterModel.broker(brokerId); | ||
| List<Disk> remainingDisks = new ArrayList<>(); | ||
| currentBroker.disks().stream().filter(disk -> !logDirs.contains(disk.logDir())).forEach(remainingDisks::add); | ||
|
|
||
| int step = 0; | ||
| int size = remainingDisks.size(); | ||
| while (!logDirs.isEmpty()) { | ||
| String logDirToRemove = (String) logDirs.toArray()[0]; | ||
|
|
||
| Set<Replica> replicasToMove = currentBroker.disk(logDirToRemove).replicas(); | ||
| while (!replicasToMove.isEmpty()) { | ||
| Replica replica = (Replica) replicasToMove.toArray()[0]; | ||
| clusterModel.relocateReplica(replica.topicPartition(), brokerId, remainingDisks.get(step % size).logDir()); | ||
| } | ||
|
|
||
| logDirs.remove(logDirToRemove); | ||
| step++; | ||
| } | ||
| } | ||
|
|
||
| return true; | ||
| } | ||
|
|
||
| @Override | ||
| public ActionAcceptance actionAcceptance(BalancingAction action, ClusterModel clusterModel) { | ||
| return ACCEPT; | ||
| } | ||
|
|
||
| @Override | ||
| public ClusterModelStatsComparator clusterModelStatsComparator() { | ||
| return new ClusterModelStatsComparator() { | ||
| @Override | ||
| public int compare(ClusterModelStats stats1, ClusterModelStats stats2) { | ||
| return 0; | ||
| } | ||
|
|
||
| @Override | ||
| public String explainLastComparison() { | ||
| return String.format("Comparison for the %s is irrelevant.", name()); | ||
| } | ||
| }; | ||
| } | ||
|
|
||
| @Override | ||
| public ModelCompletenessRequirements clusterModelCompletenessRequirements() { | ||
| return new ModelCompletenessRequirements(MIN_NUM_VALID_WINDOWS_FOR_SELF_HEALING, 0, true); | ||
| } | ||
|
|
||
| @Override | ||
| public String name() { | ||
| return DiskRemovalGoal.class.getSimpleName(); | ||
| } | ||
|
|
||
| @Override | ||
| public void finish() { | ||
|
|
||
| } | ||
|
|
||
| @Override | ||
| public boolean isHardGoal() { | ||
| return false; | ||
| } | ||
|
|
||
| @Override | ||
| public ProvisionStatus provisionStatus() { | ||
| // Provision status computation is not relevant to PLE goal. | ||
| return provisionResponse().status(); | ||
| } | ||
|
|
||
| @Override | ||
| public ProvisionResponse provisionResponse() { | ||
| return _provisionResponse; | ||
| } | ||
|
|
||
| @Override | ||
| public void configure(Map<String, ?> configs) { | ||
|
|
||
| } | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
| @@ -0,0 +1,46 @@ | ||||||
| /* | ||||||
| * Copyright 2018 LinkedIn Corp. Licensed under the BSD 2-Clause License (the "License"). See License in the project root for license information. | ||||||
|
There was a problem hiding this comment.
Suggested change
|
||||||
| */ | ||||||
|
|
||||||
| package com.linkedin.kafka.cruisecontrol.servlet.handler.async; | ||||||
|
|
||||||
| import com.linkedin.kafka.cruisecontrol.servlet.handler.async.runnable.OperationFuture; | ||||||
| import com.linkedin.kafka.cruisecontrol.servlet.handler.async.runnable.RemoveDisksRunnable; | ||||||
| import com.linkedin.kafka.cruisecontrol.servlet.parameters.RemoveDisksParameters; | ||||||
| import java.util.Map; | ||||||
|
|
||||||
| import static com.linkedin.cruisecontrol.common.utils.Utils.validateNotNull; | ||||||
| import static com.linkedin.kafka.cruisecontrol.servlet.parameters.ParameterUtils.REMOVE_DISKS_PARAMETER_OBJECT_CONFIG; | ||||||
|
|
||||||
| public class RemoveDisksRequest extends AbstractAsyncRequest { | ||||||
| protected RemoveDisksParameters _parameters; | ||||||
|
|
||||||
| public RemoveDisksRequest() { | ||||||
| super(); | ||||||
| } | ||||||
|
|
||||||
| @Override | ||||||
| protected OperationFuture handle(String uuid) { | ||||||
| OperationFuture future = new OperationFuture("Remove disks"); | ||||||
| pending(future.operationProgress()); | ||||||
| _asyncKafkaCruiseControl.sessionExecutor().submit(new RemoveDisksRunnable(_asyncKafkaCruiseControl, future, _parameters, uuid)); | ||||||
| return future; | ||||||
| } | ||||||
|
|
||||||
| @Override | ||||||
| public RemoveDisksParameters parameters() { | ||||||
| return _parameters; | ||||||
| } | ||||||
|
|
||||||
| @Override | ||||||
| public String name() { | ||||||
| return RemoveDisksRequest.class.getSimpleName(); | ||||||
| } | ||||||
|
|
||||||
| @Override | ||||||
| public void configure(Map<String, ?> configs) { | ||||||
| super.configure(configs); | ||||||
| _parameters = (RemoveDisksParameters) validateNotNull(configs.get(REMOVE_DISKS_PARAMETER_OBJECT_CONFIG), | ||||||
| "Parameter configuration is missing from the request."); | ||||||
| } | ||||||
| } | ||||||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Would the above suggestion work? What is different:
while (!logDirs.isEmpty())and(String) logDirs.toArray()[0]to iterate through a set looked very odd (same with the nestedreplicasToMoveloop)logDirs,steporsize)It is optional, you decide.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Refactored the code to improve readability. As a note, there are 2 nested loops there. The outside loop is refactored to use a for each iterator, however the inside loop calls a function which changes its iterator. This is why it is treated as a loop over a queue, so that it can change during the loop.