@Test public void replicaAllocationTest() throws Exception { String fullHostID = "full"; HostState fullHost = installHostStateWithUUID(fullHostID, spawn, true, new HostLocation("a", "", "")); fullHost.setUsed(new HostCapacity(0, 0, 0, 9998)); fullHost.setMax(new HostCapacity(0, 0, 0, 10_000)); int numLightHosts = 9; ArrayList<String> hostIDs = new ArrayList<>(numLightHosts + 1); hostIDs.add(fullHostID); String[] dataCenterIds = {"a", "b", "c"}; for (int i = 0; i < numLightHosts; i++) { String hostID = "light" + i; hostIDs.add(hostID); HostState lightHost = installHostStateWithUUID(hostID, spawn, true, new HostLocation(dataCenterIds[i % dataCenterIds.length], "", "" )); lightHost.setUsed(new HostCapacity(0, 0, 0, 20_000_000_000L)); lightHost.setMax(new HostCapacity(0, 0, 0, 100_000_000_000L)); } waitForAllUpHosts(); int numReplicas = 5; Job job = createJobAndUpdateHosts(spawn, numLightHosts + 1, hostIDs, now, 1000, 0); job.setReplicas(numReplicas); Map<Integer, List<String>> assignments = bal.getAssignmentsForNewReplicas(job); HashSet<String> usedHosts = new HashSet<>(numLightHosts); for (List<String> targets : assignments.values()) { assertEquals("should make one replica per task", numReplicas, targets.size()); assertTrue("should not put replicas on full host", !targets.contains(fullHostID)); usedHosts.addAll(targets); } assertTrue("should use many light hosts", numLightHosts > .75 * usedHosts.size()); for(JobTask task : job.getCopyOfTasks()) { List<String> replicaAssignments = assignments.get(task.getTaskID()); assertTrue("Should not put replica on the same host as the task", !replicaAssignments.contains(task.getHostUUID())); } for(JobTask task : job.getCopyOfTasks()) { List<String> replicaAssignments = assignments.get(task.getTaskID()); HashSet<HostLocation> locations = new HashSet<>(replicaAssignments.size()); for(String replica : replicaAssignments) { locations.add(hostManager.getHostState(replica).getHostLocation()); } locations.remove(hostManager.getHostState(task.getHostUUID()).getHostLocation()); assertTrue(!locations.isEmpty()); } }
public Map<Integer, List<String>> getAssignmentsForNewReplicas(IJob job) { return getAssignmentsForNewReplicas(job, -1); }
SpawnBalancer implements Codable, AutoCloseable { public Map<Integer, List<String>> getAssignmentsForNewReplicas(IJob job) { return getAssignmentsForNewReplicas(job, -1); } }
SpawnBalancer implements Codable, AutoCloseable { public Map<Integer, List<String>> getAssignmentsForNewReplicas(IJob job) { return getAssignmentsForNewReplicas(job, -1); } SpawnBalancer(Spawn spawn, HostManager hostManager); }
SpawnBalancer implements Codable, AutoCloseable { public Map<Integer, List<String>> getAssignmentsForNewReplicas(IJob job) { return getAssignmentsForNewReplicas(job, -1); } 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 { public Map<Integer, List<String>> getAssignmentsForNewReplicas(IJob job) { return getAssignmentsForNewReplicas(job, -1); } 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);  }