@Test public void toggleHostTest() throws Exception { zkClient.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/minion/up"); zkClient.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/minion/dead"); HostState tmpDisableHostState = createHostState("tmp"); tmpDisableHostState.setDisabled(true); assertTrue("disabled host states should not be able to run tasks", !tmpDisableHostState.canMirrorTasks()); try (Spawn spawn = Configs.newDefault(Spawn.class)) { HostState toggleHost = createHostState("toggle"); HostState otherHost = createHostState("other"); spawn.hostManager.updateHostState(toggleHost); spawn.hostManager.updateHostState(otherHost); spawn.toggleHosts(toggleHost.getHost(), true); assertTrue("toggled host should be disabled", spawn.hostManager.getHostState(toggleHost.getHostUuid()) .isDisabled()); assertTrue("other host should not be disabled", !spawn.hostManager.getHostState(otherHost.getHostUuid()) .isDisabled()); spawn.toggleHosts(toggleHost.getHost(), false); assertTrue("toggled host should now be re-enabled", !spawn.hostManager.getHostState(toggleHost.getHostUuid()).isDisabled()); } }
public void toggleHosts(String hosts, boolean disable) { if (hosts != null) { String[] hostsArray = hosts.split(","); for (String host : hostsArray) { if (host.isEmpty()) { continue; } boolean changed; changed = disable ? spawnState.disabledHosts.add(host) : spawnState.disabledHosts.remove(host); if (changed) { updateToggledHosts(host, disable); } } writeState(); } }
Spawn implements Codable, AutoCloseable { public void toggleHosts(String hosts, boolean disable) { if (hosts != null) { String[] hostsArray = hosts.split(","); for (String host : hostsArray) { if (host.isEmpty()) { continue; } boolean changed; changed = disable ? spawnState.disabledHosts.add(host) : spawnState.disabledHosts.remove(host); if (changed) { updateToggledHosts(host, disable); } } writeState(); } } }
Spawn implements Codable, AutoCloseable { public void toggleHosts(String hosts, boolean disable) { if (hosts != null) { String[] hostsArray = hosts.split(","); for (String host : hostsArray) { if (host.isEmpty()) { continue; } boolean changed; changed = disable ? spawnState.disabledHosts.add(host) : spawnState.disabledHosts.remove(host); if (changed) { updateToggledHosts(host, disable); } } writeState(); } } @JsonCreator private  Spawn(@JsonProperty("debug") String debug,
                               @JsonProperty(value = "queryPort", required = true) int queryPort,
                               @JsonProperty("queryHttpHost") String queryHttpHost,
                               @JsonProperty("httpHost") String httpHost,
                               @JsonProperty("meshHttpPort") String meshHttpPort,
                               @JsonProperty("dataDir") File dataDir,
                               @Nonnull @JsonProperty("stateFile") File stateFile,
                               @Nonnull @JsonProperty("expandKickExecutor") ExecutorService expandKickExecutor,
                               @Nonnull @JsonProperty("scheduledExecutor") ScheduledExecutorService scheduledExecutor,
                               @Time(MILLISECONDS) @JsonProperty(value = "taskQueueDrainInterval",
                                       required = true) int taskQueueDrainInterval,
                               @Time(MILLISECONDS) @JsonProperty(value = "hostStatusRequestInterval",
                                       required = true) int hostStatusRequestInterval,
                               @Time(MILLISECONDS) @JsonProperty(value = "queueKickInterval",
                                       required = true) int queueKickInterval,
                               @Time(MILLISECONDS) @JsonProperty("jobTaskUpdateHeartbeatInterval")
                                       int jobTaskUpdateHeartbeatInterval,
                               @Nullable @JsonProperty("structuredLogDir") File structuredLogDir,
                               @Nullable @JsonProperty("jobStore") JobStore jobStore,
                               @Nullable @JsonProperty("queueType") String queueType,
                               @Nullable @JacksonInject CuratorFramework providedZkClient,
                               @Nonnull @JsonProperty(value = "permissionsManager",
                                       required = true) PermissionsManager permissionsManager,
                               @Nonnull @JsonProperty(value = "jobDefaults",
                                       required = true) JobDefaults jobDefaults,
                               @Bytes @JsonProperty(value = "datastoreCacheSize") long datastoreCacheSize,
                               @Nonnull @JsonProperty(value = "groupManager", required = true) GroupManager groupManager); }
Spawn implements Codable, AutoCloseable { public void toggleHosts(String hosts, boolean disable) { if (hosts != null) { String[] hostsArray = hosts.split(","); for (String host : hostsArray) { if (host.isEmpty()) { continue; } boolean changed; changed = disable ? spawnState.disabledHosts.add(host) : spawnState.disabledHosts.remove(host); if (changed) { updateToggledHosts(host, disable); } } writeState(); } } @JsonCreator private  Spawn(@JsonProperty("debug") String debug,
                               @JsonProperty(value = "queryPort", required = true) int queryPort,
                               @JsonProperty("queryHttpHost") String queryHttpHost,
                               @JsonProperty("httpHost") String httpHost,
                               @JsonProperty("meshHttpPort") String meshHttpPort,
                               @JsonProperty("dataDir") File dataDir,
                               @Nonnull @JsonProperty("stateFile") File stateFile,
                               @Nonnull @JsonProperty("expandKickExecutor") ExecutorService expandKickExecutor,
                               @Nonnull @JsonProperty("scheduledExecutor") ScheduledExecutorService scheduledExecutor,
                               @Time(MILLISECONDS) @JsonProperty(value = "taskQueueDrainInterval",
                                       required = true) int taskQueueDrainInterval,
                               @Time(MILLISECONDS) @JsonProperty(value = "hostStatusRequestInterval",
                                       required = true) int hostStatusRequestInterval,
                               @Time(MILLISECONDS) @JsonProperty(value = "queueKickInterval",
                                       required = true) int queueKickInterval,
                               @Time(MILLISECONDS) @JsonProperty("jobTaskUpdateHeartbeatInterval")
                                       int jobTaskUpdateHeartbeatInterval,
                               @Nullable @JsonProperty("structuredLogDir") File structuredLogDir,
                               @Nullable @JsonProperty("jobStore") JobStore jobStore,
                               @Nullable @JsonProperty("queueType") String queueType,
                               @Nullable @JacksonInject CuratorFramework providedZkClient,
                               @Nonnull @JsonProperty(value = "permissionsManager",
                                       required = true) PermissionsManager permissionsManager,
                               @Nonnull @JsonProperty(value = "jobDefaults",
                                       required = true) JobDefaults jobDefaults,
                               @Bytes @JsonProperty(value = "datastoreCacheSize") long datastoreCacheSize,
                               @Nonnull @JsonProperty(value = "groupManager", required = true) GroupManager groupManager); static void main(String... args); @Nonnull HostFailWorker getHostFailWorker(); SpawnBalancer getSpawnBalancer(); @Nonnull AliasManager getAliasManager(); @Nonnull JobAlertManager getJobAlertManager(); @Nonnull JobEntityManager<JobMacro> getJobMacroManager(); @Nonnull SystemManager getSystemManager(); void acquireJobLock(); void releaseJobLock(); String getUuid(); void setSpawnMQ(SpawnMQ spawnMQ); @VisibleForTesting void writeSpawnQueue(); JobExpander getJobExpander(); PipedInputStream getSearchResultStream(SearchOptions searchOptions); ClientEventListener getClientEventListener(String id); @Nullable HostState markHostStateDead(String hostUUID); void sendHostUpdateEvent(HostState state); @Nullable JSONObject getHostStateUpdateEvent(HostState state); Collection<String> listAvailableHostIds(); void requestHostsUpdate(); DirectedGraph<String> getJobDependencies(); Map<ScheduledBackupType, SortedSet<Long>> getJobBackups(String jobUUID, int nodeId); boolean isSpawnMeshAvailable(); void deleteHost(String hostuuid); Collection<Job> listJobsConcurrentImmutable(); int getTaskQueuedCount(); void setJobConfig(String jobUUID, String config); JobConfigManager getJobConfigManager(); JSONArray getJobHistory(String jobId); @Nullable String getJobHistoricalConfig(String jobId, String commitId); @Nullable String diff(String jobId, String commitId); String getDeletedJobConfig(String jobId); Job createJob(String creator,
                         int taskCount,
                         Collection<String> taskHosts,
                         String minionType,
                         String command,
                         boolean defaults); @Nullable Job putJobInSpawnState(Job job); void submitConfigUpdate(String jobId, String user, @Nullable String commitMessage); @Nullable Job getJob(String jobUUID); @Nullable String getJobConfig(String jobUUID); Response synchronizeJobState(String jobUUID, String user, String token, String sudo); @Nonnull PermissionsManager getPermissionsManager(); Collection<Job> listJobs(); boolean rebalanceReplicas(Job job); boolean rebalanceReplicas(Job job, int taskID); List<JobTaskReplica> replicateTask(JobTask task, List<String> targetHosts); boolean isNewTask(JobTask task); @Nonnull JobEntityManager<JobCommand> getJobCommandManager(); List<JobTaskMoveAssignment> reallocateJob(String jobUUID, int tasksToMove); boolean swapTask(JobTask task, String replicaHostID, boolean kickOnComplete, int priority); RebalanceOutcome rebalanceHost(String hostUUID); List<JobTaskMoveAssignment> executeReallocationAssignments(@Nullable List<JobTaskMoveAssignment> assignments,
                                                                      boolean limitToAvailableSlots); boolean deleteTask(String jobUUID, String hostUuid, Integer node, boolean isReplica); @Nullable JobTask getTask(JobKey jobKey); @Nullable Job getJob(JobKey jobKey); void updateJob(@Nullable IJob ijob); void queueJobTaskUpdateEvent(IJob job); @Nullable JobTask getTask(String jobUUID, int taskID); void updateJob(@Nullable IJob ijob, boolean reviseReplicas); Set<String> getDataSources(String jobId); RebalanceOutcome rebalanceJob(String jobUUID, int tasksToMove, String user, String token, String sudo); JSONObject fixTaskDir(String jobId, int node, boolean ignoreTaskState, boolean orphansOnly); boolean resolveJobTaskDirectoryMatches(JobTask task, boolean deleteOrphansOnly); JSONArray checkTaskDirJSON(String jobId, int node); List<JobTaskDirectoryMatch> matchTaskToDirectories(JobTask task, boolean includeCorrect); boolean checkStatusForMove(String hostID); boolean prepareTaskStatesForRebalance(Job job, JobTask task, boolean isMigration); DeleteStatus forceDeleteJob(String jobUUID); DeleteStatus deleteJob(String jobUUID); void sendJobUpdateEvent(String label, IJob job); static JSONObject getJobUpdateEvent(IJob job); JSONArray getJobUpdateEventsSafely(); void sendControlMessage(CoreMessage hostMessage); void startJob(String jobUUID, int priority); void addToTaskQueue(JobKey jobKey, int priority, boolean toHead); void sendTaskQueueUpdateEvent(); void stopJob(String jobUUID); void killJob(String jobUUID); void startTask(String jobUUID, int taskID, int priority, boolean toQueueHead); void stopTask(String jobUUID, int taskID); void killTask(String jobUUID, int taskID); boolean moveTask(JobKey jobKey, String sourceUUID, String targetUUID); boolean revertJobOrTask(String jobUUID,
                                   String user,
                                   String token,
                                   String sudo,
                                   int taskID,
                                   String backupType,
                                   int rev,
                                   long time); void handleTaskError(Job job, JobTask task, int exitCode); void handleRebalanceFinish(IJob job, JobTask task, StatusTaskEnd update); JobMacro createJobHostMacro(String job, int port); void saveAllJobs(); void sendClusterQuiesceEvent(String username); int getLastQueueSize(); @Override void close(); boolean scheduleTask(Job job, JobTask task, int priority); boolean kickOnExistingHosts(Job job,
                                       JobTask task,
                                       long timeOnQueue,
                                       int priority,
                                       boolean allowSwap); void kickJobsOnQueue(); void toggleHosts(String hosts, boolean disable); void updateToggledHosts(String id, boolean disable); @Nonnull SpawnState getSpawnState(); @Nonnull JobDefaults getJobDefaults(); @Nonnull SpawnDataStore getSpawnDataStore(); }
Spawn implements Codable, AutoCloseable { public void toggleHosts(String hosts, boolean disable) { if (hosts != null) { String[] hostsArray = hosts.split(","); for (String host : hostsArray) { if (host.isEmpty()) { continue; } boolean changed; changed = disable ? spawnState.disabledHosts.add(host) : spawnState.disabledHosts.remove(host); if (changed) { updateToggledHosts(host, disable); } } writeState(); } } @JsonCreator private  Spawn(@JsonProperty("debug") String debug,
                               @JsonProperty(value = "queryPort", required = true) int queryPort,
                               @JsonProperty("queryHttpHost") String queryHttpHost,
                               @JsonProperty("httpHost") String httpHost,
                               @JsonProperty("meshHttpPort") String meshHttpPort,
                               @JsonProperty("dataDir") File dataDir,
                               @Nonnull @JsonProperty("stateFile") File stateFile,
                               @Nonnull @JsonProperty("expandKickExecutor") ExecutorService expandKickExecutor,
                               @Nonnull @JsonProperty("scheduledExecutor") ScheduledExecutorService scheduledExecutor,
                               @Time(MILLISECONDS) @JsonProperty(value = "taskQueueDrainInterval",
                                       required = true) int taskQueueDrainInterval,
                               @Time(MILLISECONDS) @JsonProperty(value = "hostStatusRequestInterval",
                                       required = true) int hostStatusRequestInterval,
                               @Time(MILLISECONDS) @JsonProperty(value = "queueKickInterval",
                                       required = true) int queueKickInterval,
                               @Time(MILLISECONDS) @JsonProperty("jobTaskUpdateHeartbeatInterval")
                                       int jobTaskUpdateHeartbeatInterval,
                               @Nullable @JsonProperty("structuredLogDir") File structuredLogDir,
                               @Nullable @JsonProperty("jobStore") JobStore jobStore,
                               @Nullable @JsonProperty("queueType") String queueType,
                               @Nullable @JacksonInject CuratorFramework providedZkClient,
                               @Nonnull @JsonProperty(value = "permissionsManager",
                                       required = true) PermissionsManager permissionsManager,
                               @Nonnull @JsonProperty(value = "jobDefaults",
                                       required = true) JobDefaults jobDefaults,
                               @Bytes @JsonProperty(value = "datastoreCacheSize") long datastoreCacheSize,
                               @Nonnull @JsonProperty(value = "groupManager", required = true) GroupManager groupManager); static void main(String... args); @Nonnull HostFailWorker getHostFailWorker(); SpawnBalancer getSpawnBalancer(); @Nonnull AliasManager getAliasManager(); @Nonnull JobAlertManager getJobAlertManager(); @Nonnull JobEntityManager<JobMacro> getJobMacroManager(); @Nonnull SystemManager getSystemManager(); void acquireJobLock(); void releaseJobLock(); String getUuid(); void setSpawnMQ(SpawnMQ spawnMQ); @VisibleForTesting void writeSpawnQueue(); JobExpander getJobExpander(); PipedInputStream getSearchResultStream(SearchOptions searchOptions); ClientEventListener getClientEventListener(String id); @Nullable HostState markHostStateDead(String hostUUID); void sendHostUpdateEvent(HostState state); @Nullable JSONObject getHostStateUpdateEvent(HostState state); Collection<String> listAvailableHostIds(); void requestHostsUpdate(); DirectedGraph<String> getJobDependencies(); Map<ScheduledBackupType, SortedSet<Long>> getJobBackups(String jobUUID, int nodeId); boolean isSpawnMeshAvailable(); void deleteHost(String hostuuid); Collection<Job> listJobsConcurrentImmutable(); int getTaskQueuedCount(); void setJobConfig(String jobUUID, String config); JobConfigManager getJobConfigManager(); JSONArray getJobHistory(String jobId); @Nullable String getJobHistoricalConfig(String jobId, String commitId); @Nullable String diff(String jobId, String commitId); String getDeletedJobConfig(String jobId); Job createJob(String creator,
                         int taskCount,
                         Collection<String> taskHosts,
                         String minionType,
                         String command,
                         boolean defaults); @Nullable Job putJobInSpawnState(Job job); void submitConfigUpdate(String jobId, String user, @Nullable String commitMessage); @Nullable Job getJob(String jobUUID); @Nullable String getJobConfig(String jobUUID); Response synchronizeJobState(String jobUUID, String user, String token, String sudo); @Nonnull PermissionsManager getPermissionsManager(); Collection<Job> listJobs(); boolean rebalanceReplicas(Job job); boolean rebalanceReplicas(Job job, int taskID); List<JobTaskReplica> replicateTask(JobTask task, List<String> targetHosts); boolean isNewTask(JobTask task); @Nonnull JobEntityManager<JobCommand> getJobCommandManager(); List<JobTaskMoveAssignment> reallocateJob(String jobUUID, int tasksToMove); boolean swapTask(JobTask task, String replicaHostID, boolean kickOnComplete, int priority); RebalanceOutcome rebalanceHost(String hostUUID); List<JobTaskMoveAssignment> executeReallocationAssignments(@Nullable List<JobTaskMoveAssignment> assignments,
                                                                      boolean limitToAvailableSlots); boolean deleteTask(String jobUUID, String hostUuid, Integer node, boolean isReplica); @Nullable JobTask getTask(JobKey jobKey); @Nullable Job getJob(JobKey jobKey); void updateJob(@Nullable IJob ijob); void queueJobTaskUpdateEvent(IJob job); @Nullable JobTask getTask(String jobUUID, int taskID); void updateJob(@Nullable IJob ijob, boolean reviseReplicas); Set<String> getDataSources(String jobId); RebalanceOutcome rebalanceJob(String jobUUID, int tasksToMove, String user, String token, String sudo); JSONObject fixTaskDir(String jobId, int node, boolean ignoreTaskState, boolean orphansOnly); boolean resolveJobTaskDirectoryMatches(JobTask task, boolean deleteOrphansOnly); JSONArray checkTaskDirJSON(String jobId, int node); List<JobTaskDirectoryMatch> matchTaskToDirectories(JobTask task, boolean includeCorrect); boolean checkStatusForMove(String hostID); boolean prepareTaskStatesForRebalance(Job job, JobTask task, boolean isMigration); DeleteStatus forceDeleteJob(String jobUUID); DeleteStatus deleteJob(String jobUUID); void sendJobUpdateEvent(String label, IJob job); static JSONObject getJobUpdateEvent(IJob job); JSONArray getJobUpdateEventsSafely(); void sendControlMessage(CoreMessage hostMessage); void startJob(String jobUUID, int priority); void addToTaskQueue(JobKey jobKey, int priority, boolean toHead); void sendTaskQueueUpdateEvent(); void stopJob(String jobUUID); void killJob(String jobUUID); void startTask(String jobUUID, int taskID, int priority, boolean toQueueHead); void stopTask(String jobUUID, int taskID); void killTask(String jobUUID, int taskID); boolean moveTask(JobKey jobKey, String sourceUUID, String targetUUID); boolean revertJobOrTask(String jobUUID,
                                   String user,
                                   String token,
                                   String sudo,
                                   int taskID,
                                   String backupType,
                                   int rev,
                                   long time); void handleTaskError(Job job, JobTask task, int exitCode); void handleRebalanceFinish(IJob job, JobTask task, StatusTaskEnd update); JobMacro createJobHostMacro(String job, int port); void saveAllJobs(); void sendClusterQuiesceEvent(String username); int getLastQueueSize(); @Override void close(); boolean scheduleTask(Job job, JobTask task, int priority); boolean kickOnExistingHosts(Job job,
                                       JobTask task,
                                       long timeOnQueue,
                                       int priority,
                                       boolean allowSwap); void kickJobsOnQueue(); void toggleHosts(String hosts, boolean disable); void updateToggledHosts(String id, boolean disable); @Nonnull SpawnState getSpawnState(); @Nonnull JobDefaults getJobDefaults(); @Nonnull SpawnDataStore getSpawnDataStore(); static final long INPUT_MAX_NUMBER_OF_CHARACTERS; @Nonnull final HostManager hostManager; }