@Test public void assignMoreTasksToLighterHostsTest() throws Exception { String heavyHostUUID = "heavy"; String lightHost1UUID = "light1"; String lightHost2UUID = "light2"; HostState heavyHost = installHostStateWithUUID(heavyHostUUID, spawn, true, new HostLocation("", "", "")); HostState lightHost1 = installHostStateWithUUID(lightHost1UUID, spawn, true, new HostLocation("", "", "")); HostState lightHost2 = installHostStateWithUUID(lightHost2UUID, spawn, true, new HostLocation("", "", "")); waitForAllUpHosts(); Job weightJob = createJobAndUpdateHosts(spawn, 10, Arrays.asList(heavyHostUUID), now, 1000, 0); bal.updateAggregateStatistics(hostManager.listHostStatus(null)); Job otherJob = createJobAndUpdateHosts(spawn, 5, Arrays.asList(heavyHostUUID, lightHost1UUID, lightHost2UUID), now, 500, 0); assertEquals("should put two tasks on lightHost1", 2, numTasksOnHost(otherJob, lightHost1UUID)); assertEquals("should put two tasks on lightHost2", 2, numTasksOnHost(otherJob, lightHost2UUID)); assertEquals("should put one task on heavyHost", 1, numTasksOnHost(otherJob, heavyHostUUID)); }
protected void updateAggregateStatistics(List<HostState> hosts) { aggregateStatisticsLock.lock(); try { updateActiveJobIDs(); double maxMeanActive = -1; double maxDiskPercentUsed = -1; double minDiskPercentUsed = 1; double sumDiskPercentUsed = 0; double maxTaskPercent = 0; double minTaskPercent = 1; double sumTaskPercent = 0; for (HostState host : hosts) { maxMeanActive = Math.max(maxMeanActive, host.getMeanActiveTasks()); double diskPercentUsed = host.getDiskUsedPercent(); sumDiskPercentUsed += diskPercentUsed; maxDiskPercentUsed = Math.max(diskPercentUsed, maxDiskPercentUsed); minDiskPercentUsed = Math.min(diskPercentUsed, minDiskPercentUsed); double taskPercent = host.getMeanActiveTasks(); sumTaskPercent += taskPercent; maxTaskPercent = Math.max(taskPercent, maxTaskPercent); minTaskPercent = Math.min(taskPercent, minTaskPercent); } int numScores = hosts.size(); double avgDiskPercentUsed = sumDiskPercentUsed / (double) numScores; double sumDiskPercentUsedDiff = 0; double avgTaskPercent = sumTaskPercent / (double) numScores; double sumTaskPercentDiff = 0; for (HostState host : hosts) { HostScore score = calculateHostScore(host, maxMeanActive, maxDiskPercentUsed); cachedHostScores.put(host.getHostUuid(), score); double diskDiff = Math.abs(avgDiskPercentUsed - host.getDiskUsedPercent()); sumDiskPercentUsedDiff += diskDiff; double taskDiff = score.getScoreValue(false); sumTaskPercentDiff += Math.abs(avgTaskPercent - taskDiff); } avgDiskPercentUsedDiff = sumDiskPercentUsedDiff / (double) numScores; avgTaskPercentDiff = sumTaskPercentDiff / (double) numScores; minDiskPercentUsedDiff = avgDiskPercentUsed - minDiskPercentUsed; maxDiskPercentUsedDiff = maxDiskPercentUsed - avgDiskPercentUsed; minTaskPercentDiff = avgTaskPercent - minTaskPercent; maxTaskPercentDiff = maxTaskPercent - avgTaskPercent; } finally { aggregateStatisticsLock.unlock(); } }
SpawnBalancer implements Codable, AutoCloseable { protected void updateAggregateStatistics(List<HostState> hosts) { aggregateStatisticsLock.lock(); try { updateActiveJobIDs(); double maxMeanActive = -1; double maxDiskPercentUsed = -1; double minDiskPercentUsed = 1; double sumDiskPercentUsed = 0; double maxTaskPercent = 0; double minTaskPercent = 1; double sumTaskPercent = 0; for (HostState host : hosts) { maxMeanActive = Math.max(maxMeanActive, host.getMeanActiveTasks()); double diskPercentUsed = host.getDiskUsedPercent(); sumDiskPercentUsed += diskPercentUsed; maxDiskPercentUsed = Math.max(diskPercentUsed, maxDiskPercentUsed); minDiskPercentUsed = Math.min(diskPercentUsed, minDiskPercentUsed); double taskPercent = host.getMeanActiveTasks(); sumTaskPercent += taskPercent; maxTaskPercent = Math.max(taskPercent, maxTaskPercent); minTaskPercent = Math.min(taskPercent, minTaskPercent); } int numScores = hosts.size(); double avgDiskPercentUsed = sumDiskPercentUsed / (double) numScores; double sumDiskPercentUsedDiff = 0; double avgTaskPercent = sumTaskPercent / (double) numScores; double sumTaskPercentDiff = 0; for (HostState host : hosts) { HostScore score = calculateHostScore(host, maxMeanActive, maxDiskPercentUsed); cachedHostScores.put(host.getHostUuid(), score); double diskDiff = Math.abs(avgDiskPercentUsed - host.getDiskUsedPercent()); sumDiskPercentUsedDiff += diskDiff; double taskDiff = score.getScoreValue(false); sumTaskPercentDiff += Math.abs(avgTaskPercent - taskDiff); } avgDiskPercentUsedDiff = sumDiskPercentUsedDiff / (double) numScores; avgTaskPercentDiff = sumTaskPercentDiff / (double) numScores; minDiskPercentUsedDiff = avgDiskPercentUsed - minDiskPercentUsed; maxDiskPercentUsedDiff = maxDiskPercentUsed - avgDiskPercentUsed; minTaskPercentDiff = avgTaskPercent - minTaskPercent; maxTaskPercentDiff = maxTaskPercent - avgTaskPercent; } finally { aggregateStatisticsLock.unlock(); } } }
SpawnBalancer implements Codable, AutoCloseable { protected void updateAggregateStatistics(List<HostState> hosts) { aggregateStatisticsLock.lock(); try { updateActiveJobIDs(); double maxMeanActive = -1; double maxDiskPercentUsed = -1; double minDiskPercentUsed = 1; double sumDiskPercentUsed = 0; double maxTaskPercent = 0; double minTaskPercent = 1; double sumTaskPercent = 0; for (HostState host : hosts) { maxMeanActive = Math.max(maxMeanActive, host.getMeanActiveTasks()); double diskPercentUsed = host.getDiskUsedPercent(); sumDiskPercentUsed += diskPercentUsed; maxDiskPercentUsed = Math.max(diskPercentUsed, maxDiskPercentUsed); minDiskPercentUsed = Math.min(diskPercentUsed, minDiskPercentUsed); double taskPercent = host.getMeanActiveTasks(); sumTaskPercent += taskPercent; maxTaskPercent = Math.max(taskPercent, maxTaskPercent); minTaskPercent = Math.min(taskPercent, minTaskPercent); } int numScores = hosts.size(); double avgDiskPercentUsed = sumDiskPercentUsed / (double) numScores; double sumDiskPercentUsedDiff = 0; double avgTaskPercent = sumTaskPercent / (double) numScores; double sumTaskPercentDiff = 0; for (HostState host : hosts) { HostScore score = calculateHostScore(host, maxMeanActive, maxDiskPercentUsed); cachedHostScores.put(host.getHostUuid(), score); double diskDiff = Math.abs(avgDiskPercentUsed - host.getDiskUsedPercent()); sumDiskPercentUsedDiff += diskDiff; double taskDiff = score.getScoreValue(false); sumTaskPercentDiff += Math.abs(avgTaskPercent - taskDiff); } avgDiskPercentUsedDiff = sumDiskPercentUsedDiff / (double) numScores; avgTaskPercentDiff = sumTaskPercentDiff / (double) numScores; minDiskPercentUsedDiff = avgDiskPercentUsed - minDiskPercentUsed; maxDiskPercentUsedDiff = maxDiskPercentUsed - avgDiskPercentUsed; minTaskPercentDiff = avgTaskPercent - minTaskPercent; maxTaskPercentDiff = maxTaskPercent - avgTaskPercent; } finally { aggregateStatisticsLock.unlock(); } } SpawnBalancer(Spawn spawn, HostManager hostManager); }
SpawnBalancer implements Codable, AutoCloseable { protected void updateAggregateStatistics(List<HostState> hosts) { aggregateStatisticsLock.lock(); try { updateActiveJobIDs(); double maxMeanActive = -1; double maxDiskPercentUsed = -1; double minDiskPercentUsed = 1; double sumDiskPercentUsed = 0; double maxTaskPercent = 0; double minTaskPercent = 1; double sumTaskPercent = 0; for (HostState host : hosts) { maxMeanActive = Math.max(maxMeanActive, host.getMeanActiveTasks()); double diskPercentUsed = host.getDiskUsedPercent(); sumDiskPercentUsed += diskPercentUsed; maxDiskPercentUsed = Math.max(diskPercentUsed, maxDiskPercentUsed); minDiskPercentUsed = Math.min(diskPercentUsed, minDiskPercentUsed); double taskPercent = host.getMeanActiveTasks(); sumTaskPercent += taskPercent; maxTaskPercent = Math.max(taskPercent, maxTaskPercent); minTaskPercent = Math.min(taskPercent, minTaskPercent); } int numScores = hosts.size(); double avgDiskPercentUsed = sumDiskPercentUsed / (double) numScores; double sumDiskPercentUsedDiff = 0; double avgTaskPercent = sumTaskPercent / (double) numScores; double sumTaskPercentDiff = 0; for (HostState host : hosts) { HostScore score = calculateHostScore(host, maxMeanActive, maxDiskPercentUsed); cachedHostScores.put(host.getHostUuid(), score); double diskDiff = Math.abs(avgDiskPercentUsed - host.getDiskUsedPercent()); sumDiskPercentUsedDiff += diskDiff; double taskDiff = score.getScoreValue(false); sumTaskPercentDiff += Math.abs(avgTaskPercent - taskDiff); } avgDiskPercentUsedDiff = sumDiskPercentUsedDiff / (double) numScores; avgTaskPercentDiff = sumTaskPercentDiff / (double) numScores; minDiskPercentUsedDiff = avgDiskPercentUsed - minDiskPercentUsed; maxDiskPercentUsedDiff = maxDiskPercentUsed - avgDiskPercentUsed; minTaskPercentDiff = avgTaskPercent - minTaskPercent; maxTaskPercentDiff = maxTaskPercent - avgTaskPercent; } finally { aggregateStatisticsLock.unlock(); } } SpawnBalancer(Spawn spawn, HostManager hostManager); double getHostScoreCached(String hostId); synchronized void startAutobalanceTask(); @Override void close(); void saveConfigToDataStore(); void startTaskSizePolling(); Map<JobTask, String> assignTasksFromMultipleJobsToHosts(Collection<JobTask> tasks,
                                                                   Collection<HostState> hosts); Map<HostState, Double> generateHostStateScoreMap(Collection<HostState> hosts); boolean canReceiveNewTasks(HostState host); List<JobTaskMoveAssignment> pushTasksOffHostForFilesystemOkayFailure(HostState host, int moveLimit); long getTaskTrueSize(JobTask task); static boolean isInMovableState(JobTask task); boolean okayToAutobalance(); @Nullable List<JobTaskMoveAssignment> getAssignmentsForAutoBalance(RebalanceType type,
                                                                              RebalanceWeight weight); List<Job> getJobsToAutobalance(List<HostState> hosts); Map<Integer, List<String>> getAssignmentsForNewReplicas(IJob job); Map<Integer, List<String>> getAssignmentsForNewReplicas(IJob job, int taskID); void requestJobSizeUpdate(String jobId, int taskId); SpawnBalancerConfig getConfig(); void setConfig(SpawnBalancerConfig config); void clearRecentlyRebalancedHosts(); void fixTasksForFailedHost(List<HostState> hosts, String failedHost); List<JobTask> generateAssignedTasksForNewJob(String jobID, int taskCount, Collection<HostState> hosts); List<JobTaskMoveAssignment> getAssignmentsForJobReallocation(Job job,
                                                                        int tasksToMove,
                                                                        List<HostState> hosts); List<JobTaskMoveAssignment> getAssignmentsToBalanceHost(HostState host, List<HostState> hosts); List<HostState> sortHostsByActiveTasks(Collection<HostState> hosts); void removeInvalidReplicas(IJob job); boolean isDiskFull(@Nullable HostState host); @SuppressWarnings("FloatingPointEquality") @Nonnull Collection<JobTask> getTasksToMove(String fromHostUuid, String toHostUuid,
                                                       int numTasks, boolean obeyDontAutobalanceMe); boolean isTaskSpreadOutAcrossAd(@Nullable HostLocation fromHostLocation,
                                           @Nullable HostLocation toHostLocation, JobTask task); }
SpawnBalancer implements Codable, AutoCloseable { protected void updateAggregateStatistics(List<HostState> hosts) { aggregateStatisticsLock.lock(); try { updateActiveJobIDs(); double maxMeanActive = -1; double maxDiskPercentUsed = -1; double minDiskPercentUsed = 1; double sumDiskPercentUsed = 0; double maxTaskPercent = 0; double minTaskPercent = 1; double sumTaskPercent = 0; for (HostState host : hosts) { maxMeanActive = Math.max(maxMeanActive, host.getMeanActiveTasks()); double diskPercentUsed = host.getDiskUsedPercent(); sumDiskPercentUsed += diskPercentUsed; maxDiskPercentUsed = Math.max(diskPercentUsed, maxDiskPercentUsed); minDiskPercentUsed = Math.min(diskPercentUsed, minDiskPercentUsed); double taskPercent = host.getMeanActiveTasks(); sumTaskPercent += taskPercent; maxTaskPercent = Math.max(taskPercent, maxTaskPercent); minTaskPercent = Math.min(taskPercent, minTaskPercent); } int numScores = hosts.size(); double avgDiskPercentUsed = sumDiskPercentUsed / (double) numScores; double sumDiskPercentUsedDiff = 0; double avgTaskPercent = sumTaskPercent / (double) numScores; double sumTaskPercentDiff = 0; for (HostState host : hosts) { HostScore score = calculateHostScore(host, maxMeanActive, maxDiskPercentUsed); cachedHostScores.put(host.getHostUuid(), score); double diskDiff = Math.abs(avgDiskPercentUsed - host.getDiskUsedPercent()); sumDiskPercentUsedDiff += diskDiff; double taskDiff = score.getScoreValue(false); sumTaskPercentDiff += Math.abs(avgTaskPercent - taskDiff); } avgDiskPercentUsedDiff = sumDiskPercentUsedDiff / (double) numScores; avgTaskPercentDiff = sumTaskPercentDiff / (double) numScores; minDiskPercentUsedDiff = avgDiskPercentUsed - minDiskPercentUsed; maxDiskPercentUsedDiff = maxDiskPercentUsed - avgDiskPercentUsed; minTaskPercentDiff = avgTaskPercent - minTaskPercent; maxTaskPercentDiff = maxTaskPercent - avgTaskPercent; } finally { aggregateStatisticsLock.unlock(); } } SpawnBalancer(Spawn spawn, HostManager hostManager); double getHostScoreCached(String hostId); synchronized void startAutobalanceTask(); @Override void close(); void saveConfigToDataStore(); void startTaskSizePolling(); Map<JobTask, String> assignTasksFromMultipleJobsToHosts(Collection<JobTask> tasks,
                                                                   Collection<HostState> hosts); Map<HostState, Double> generateHostStateScoreMap(Collection<HostState> hosts); boolean canReceiveNewTasks(HostState host); List<JobTaskMoveAssignment> pushTasksOffHostForFilesystemOkayFailure(HostState host, int moveLimit); long getTaskTrueSize(JobTask task); static boolean isInMovableState(JobTask task); boolean okayToAutobalance(); @Nullable List<JobTaskMoveAssignment> getAssignmentsForAutoBalance(RebalanceType type,
                                                                              RebalanceWeight weight); List<Job> getJobsToAutobalance(List<HostState> hosts); Map<Integer, List<String>> getAssignmentsForNewReplicas(IJob job); Map<Integer, List<String>> getAssignmentsForNewReplicas(IJob job, int taskID); void requestJobSizeUpdate(String jobId, int taskId); SpawnBalancerConfig getConfig(); void setConfig(SpawnBalancerConfig config); void clearRecentlyRebalancedHosts(); void fixTasksForFailedHost(List<HostState> hosts, String failedHost); List<JobTask> generateAssignedTasksForNewJob(String jobID, int taskCount, Collection<HostState> hosts); List<JobTaskMoveAssignment> getAssignmentsForJobReallocation(Job job,
                                                                        int tasksToMove,
                                                                        List<HostState> hosts); List<JobTaskMoveAssignment> getAssignmentsToBalanceHost(HostState host, List<HostState> hosts); List<HostState> sortHostsByActiveTasks(Collection<HostState> hosts); void removeInvalidReplicas(IJob job); boolean isDiskFull(@Nullable HostState host); @SuppressWarnings("FloatingPointEquality") @Nonnull Collection<JobTask> getTasksToMove(String fromHostUuid, String toHostUuid,
                                                       int numTasks, boolean obeyDontAutobalanceMe); boolean isTaskSpreadOutAcrossAd(@Nullable HostLocation fromHostLocation,
                                           @Nullable HostLocation toHostLocation, JobTask task);  }