@Test public void testRequestAbortsOnlyWhenNoQueuedTaskAndSuccessFactorUnmet() throws Exception { ActionQueue aq = new ActionQueue(); Clusters fsm = mock(Clusters.class); Cluster oneClusterMock = mock(Cluster.class); Service serviceObj = mock(Service.class); ServiceComponent scomp = mock(ServiceComponent.class); ServiceComponentHost sch = mock(ServiceComponentHost.class); UnitOfWork unitOfWork = mock(UnitOfWork.class); when(fsm.getCluster(anyString())).thenReturn(oneClusterMock); when(oneClusterMock.getService(anyString())).thenReturn(serviceObj); when(serviceObj.getServiceComponent(anyString())).thenReturn(scomp); when(scomp.getServiceComponentHost(anyString())).thenReturn(sch); when(serviceObj.getCluster()).thenReturn(oneClusterMock); String host1 = "host1"; String host2 = "host2"; Host host = mock(Host.class); HashMap<String, ServiceComponentHost> hosts = new HashMap<String, ServiceComponentHost>(); hosts.put(host1, sch); hosts.put(host2, sch); when(scomp.getServiceComponentHosts()).thenReturn(hosts); when(fsm.getHost(anyString())).thenReturn(host); when(host.getState()).thenReturn(HostState.HEALTHY); when(host.getHostName()).thenReturn(host1); HostEntity hostEntity1 = new HostEntity(); HostEntity hostEntity2 = new HostEntity(); hostEntity1.setHostName(host1); hostEntity2.setHostName(host2); hostDAO.create(hostEntity1); hostDAO.create(hostEntity2); final List<Stage> stages = new ArrayList<Stage>(); long now = System.currentTimeMillis(); Stage stage = stageFactory.createNew(1, "/tmp", "cluster1", 1L, "testRequestFailureBasedOnSuccessFactor", CLUSTER_HOST_INFO, "", ""); stage.setStageId(1); addHostRoleExecutionCommand(now, stage, Role.SQOOP, Service.Type.SQOOP, RoleCommand.INSTALL, host1, "cluster1"); addHostRoleExecutionCommand(now, stage, Role.OOZIE_CLIENT, Service.Type.OOZIE, RoleCommand.INSTALL, host1, "cluster1"); addHostRoleExecutionCommand(now, stage, Role.MAPREDUCE_CLIENT, Service.Type.MAPREDUCE, RoleCommand.INSTALL, host1, "cluster1"); addHostRoleExecutionCommand(now, stage, Role.HBASE_CLIENT, Service.Type.HBASE, RoleCommand.INSTALL, host1, "cluster1"); addHostRoleExecutionCommand(now, stage, Role.GANGLIA_MONITOR, Service.Type.GANGLIA, RoleCommand.INSTALL, host1, "cluster1"); addHostRoleExecutionCommand(now, stage, Role.HBASE_CLIENT, Service.Type.HBASE, RoleCommand.INSTALL, host2, "cluster1"); addHostRoleExecutionCommand(now, stage, Role.GANGLIA_MONITOR, Service.Type.GANGLIA, RoleCommand.INSTALL, host2, "cluster1"); stages.add(stage); HostRoleStatus[] statusesAtIterOne = {HostRoleStatus.QUEUED, HostRoleStatus.QUEUED, HostRoleStatus.QUEUED, HostRoleStatus.QUEUED, HostRoleStatus.FAILED, HostRoleStatus.FAILED, HostRoleStatus.QUEUED, HostRoleStatus.QUEUED}; for (int index = 0; index < stage.getOrderedHostRoleCommands().size(); index++) { stage.getOrderedHostRoleCommands().get(index).setTaskId(index + 1); stage.getOrderedHostRoleCommands().get(index).setStatus(statusesAtIterOne[index]); } stage.setLastAttemptTime(host1, Role.SQOOP.toString(), now); stage.setLastAttemptTime(host1, Role.MAPREDUCE_CLIENT.toString(), now); stage.setLastAttemptTime(host1, Role.OOZIE_CLIENT.toString(), now); stage.setLastAttemptTime(host1, Role.GANGLIA_MONITOR.toString(), now); stage.setLastAttemptTime(host1, Role.HBASE_CLIENT.toString(), now); stage.setLastAttemptTime(host2, Role.GANGLIA_MONITOR.toString(), now); stage.setLastAttemptTime(host2, Role.HBASE_CLIENT.toString(), now); ActionDBAccessor db = mock(ActionDBAccessor.class); HostRoleCommandDAO hostRoleCommandDAOMock = mock(HostRoleCommandDAO.class); Mockito.doNothing().when(hostRoleCommandDAOMock).publishTaskCreateEvent(anyListOf(HostRoleCommand.class)); RequestEntity request = mock(RequestEntity.class); when(request.isExclusive()).thenReturn(false); when(db.getRequestEntity(anyLong())).thenReturn(request); when(db.getCommandsInProgressCount()).thenReturn(stages.size()); when(db.getStagesInProgress()).thenReturn(stages); doAnswer(new Answer<Void>() { @Override public Void answer(InvocationOnMock invocation) throws Throwable { String host = (String) invocation.getArguments()[0]; Long requestId = (Long) invocation.getArguments()[1]; Long stageId = (Long) invocation.getArguments()[2]; String role = (String) invocation.getArguments()[3]; CommandReport commandReport = (CommandReport) invocation.getArguments()[4]; for (Stage stage : stages) { if (requestId.equals(stage.getRequestId()) && stageId.equals(stage.getStageId())) { HostRoleCommand command = stage.getHostRoleCommand(host, role); command.setStatus(HostRoleStatus.valueOf(commandReport.getStatus())); } } return null; } }).when(db).updateHostRoleState(anyString(), anyLong(), anyLong(), anyString(), any(CommandReport.class)); when(db.getTask(anyLong())).thenAnswer(new Answer<Object>() { @Override public Object answer(InvocationOnMock invocation) throws Throwable { Long taskId = (Long) invocation.getArguments()[0]; for (Stage stage : stages) { for (HostRoleCommand command : stage.getOrderedHostRoleCommands()) { if (taskId.equals(command.getTaskId())) { return command; } } } return null; } }); doAnswer(new Answer<Collection<HostRoleCommandEntity>>() { @Override public Collection<HostRoleCommandEntity> answer(InvocationOnMock invocation) throws Throwable { Long requestId = (Long) invocation.getArguments()[0]; List<HostRoleCommandEntity> abortedCommands = Lists.newArrayList(); for (Stage stage : stages) { if (requestId.equals(stage.getRequestId())) { for (HostRoleCommand command : stage.getOrderedHostRoleCommands()) { if (command.getStatus() == HostRoleStatus.QUEUED || command.getStatus() == HostRoleStatus.IN_PROGRESS || command.getStatus() == HostRoleStatus.PENDING) { command.setStatus(HostRoleStatus.ABORTED); HostRoleCommandEntity hostRoleCommandEntity = command.constructNewPersistenceEntity(); hostRoleCommandEntity.setStage(stage.constructNewPersistenceEntity()); abortedCommands.add(hostRoleCommandEntity); } } } } return abortedCommands; } }).when(db).abortOperation(anyLong()); Properties properties = new Properties(); Configuration conf = new Configuration(properties); ActionScheduler scheduler = new ActionScheduler(100, 10000, db, aq, fsm, 3, new HostsMap((String) null), unitOfWork, null, conf, entityManagerProviderMock, hostRoleCommandDAOMock, (HostRoleCommandFactory)null); scheduler.doWork(); HostRoleStatus[] expectedStatusesAtIterOne = {HostRoleStatus.QUEUED, HostRoleStatus.QUEUED, HostRoleStatus.QUEUED, HostRoleStatus.QUEUED, HostRoleStatus.FAILED, HostRoleStatus.FAILED, HostRoleStatus.QUEUED, HostRoleStatus.QUEUED}; for (int index = 0; index < stage.getOrderedHostRoleCommands().size(); index++) { log.info(stage.getOrderedHostRoleCommands().get(index).toString()); Assert.assertEquals(expectedStatusesAtIterOne[index], stage.getOrderedHostRoleCommands().get(index).getStatus()); } HostRoleStatus[] statusesAtIterTwo = {HostRoleStatus.QUEUED, HostRoleStatus.QUEUED, HostRoleStatus.QUEUED, HostRoleStatus.QUEUED, HostRoleStatus.FAILED, HostRoleStatus.FAILED, HostRoleStatus.QUEUED, HostRoleStatus.COMPLETED}; for (int index = 0; index < stage.getOrderedHostRoleCommands().size(); index++) { stage.getOrderedHostRoleCommands().get(index).setStatus(statusesAtIterTwo[index]); } scheduler.doWork(); HostRoleStatus[] expectedStatusesAtIterTwo = {HostRoleStatus.QUEUED, HostRoleStatus.QUEUED, HostRoleStatus.QUEUED, HostRoleStatus.QUEUED, HostRoleStatus.FAILED, HostRoleStatus.FAILED, HostRoleStatus.QUEUED, HostRoleStatus.COMPLETED}; for (int index = 0; index < stage.getOrderedHostRoleCommands().size(); index++) { log.info(stage.getOrderedHostRoleCommands().get(index).toString()); Assert.assertEquals(expectedStatusesAtIterTwo[index], stage.getOrderedHostRoleCommands().get(index).getStatus()); } HostRoleStatus[] statusesAtIterThree = {HostRoleStatus.QUEUED, HostRoleStatus.QUEUED, HostRoleStatus.QUEUED, HostRoleStatus.QUEUED, HostRoleStatus.FAILED, HostRoleStatus.FAILED, HostRoleStatus.FAILED, HostRoleStatus.COMPLETED}; for (int index = 0; index < stage.getOrderedHostRoleCommands().size(); index++) { stage.getOrderedHostRoleCommands().get(index).setStatus(statusesAtIterThree[index]); } scheduler.doWork(); HostRoleStatus[] expectedStatusesAtIterThree = {HostRoleStatus.ABORTED, HostRoleStatus.ABORTED, HostRoleStatus.ABORTED, HostRoleStatus.ABORTED, HostRoleStatus.FAILED, HostRoleStatus.FAILED, HostRoleStatus.FAILED, HostRoleStatus.COMPLETED}; for (int index = 0; index < stage.getOrderedHostRoleCommands().size(); index++) { log.info(stage.getOrderedHostRoleCommands().get(index).toString()); Assert.assertEquals(expectedStatusesAtIterThree[index], stage.getOrderedHostRoleCommands().get(index).getStatus()); } }
public void doWork() throws AmbariException { try { unitOfWork.begin(); threadEntityManager = entityManagerProvider.get(); processCancelledRequestsList(); if (db.getCommandsInProgressCount() == 0) { if (LOG.isDebugEnabled()) { LOG.debug("There are no stages currently in progress."); } actionQueue.updateListOfHostsWithPendingTask(null); return; } Set<Long> runningRequestIds = new HashSet<Long>(); List<Stage> stages = db.getStagesInProgress(); if (LOG.isDebugEnabled()) { LOG.debug("Scheduler wakes up"); LOG.debug("Processing {} in progress stages ", stages.size()); } publishInProgressTasks(stages); if (stages.isEmpty()) { if (LOG.isDebugEnabled()) { LOG.debug("There are no stages currently in progress."); } actionQueue.updateListOfHostsWithPendingTask(null); return; } int i_stage = 0; HashSet<String> hostsWithTasks = getListOfHostsWithPendingTask(stages); actionQueue.updateListOfHostsWithPendingTask(hostsWithTasks); stages = filterParallelPerHostStages(stages); boolean exclusiveRequestIsGoing = false; for (Stage stage : stages) { i_stage++; long requestId = stage.getRequestId(); LOG.debug("==> STAGE_i = " + i_stage + "(requestId=" + requestId + ",StageId=" + stage.getStageId() + ")"); RequestEntity request = db.getRequestEntity(requestId); if (request.isExclusive()) { if (runningRequestIds.size() > 0) { LOG.debug("Stage requires exclusive execution, but other requests are already executing. Stopping for now"); break; } exclusiveRequestIsGoing = true; } if (runningRequestIds.contains(requestId)) { LOG.debug("==> We don't want to process different stages from the same request in parallel"); continue; } else { runningRequestIds.add(requestId); if (!requestsInProgress.contains(requestId)) { requestsInProgress.add(requestId); db.startRequest(requestId); } } List<ExecutionCommand> commandsToSchedule = new ArrayList<>(); Multimap<String, AgentCommand> commandsToEnqueue = ArrayListMultimap.create(); Map<String, RoleStats> roleStats = processInProgressStage(stage, commandsToSchedule, commandsToEnqueue); boolean failed = false; for (Map.Entry<String, RoleStats> entry : roleStats.entrySet()) { String role = entry.getKey(); RoleStats stats = entry.getValue(); if (LOG.isDebugEnabled()) { LOG.debug("Stats for role: {}, stats={}", role, stats); } if (stats.isRoleFailed() && !stage.isSkippable()) { LOG.warn("{} failed, request {} will be aborted", role, request.getRequestId()); failed = true; break; } } if (!failed) { failed = hasPreviousStageFailed(stage); } if (failed) { LOG.error("Operation completely failed, aborting request id: {}", stage.getRequestId()); cancelHostRoleCommands(stage.getOrderedHostRoleCommands(), FAILED_TASK_ABORT_REASONING); abortOperationsForStage(stage); return; } List<ExecutionCommand> commandsToStart = new ArrayList<ExecutionCommand>(); List<ExecutionCommand> commandsToUpdate = new ArrayList<ExecutionCommand>(); for (ExecutionCommand cmd : commandsToSchedule) { ConfigHelper.processHiddenAttribute(cmd.getConfigurations(), cmd.getConfigurationAttributes(), cmd.getRole(), false); processHostRole(stage, cmd, commandsToStart, commandsToUpdate); } LOG.debug("==> Commands to start: {}", commandsToStart.size()); LOG.debug("==> Commands to update: {}", commandsToUpdate.size()); ListMultimap<String, ServiceComponentHostEvent> eventMap = formEventMap(stage, commandsToStart); Map<ExecutionCommand, String> commandsToAbort = new HashMap<ExecutionCommand, String>(); if (!eventMap.isEmpty()) { LOG.debug("==> processing {} serviceComponentHostEvents...", eventMap.size()); Cluster cluster = clusters.getCluster(stage.getClusterName()); if (cluster != null) { Map<ServiceComponentHostEvent, String> failedEvents = cluster.processServiceComponentHostEvents(eventMap); if (failedEvents.size() > 0) { LOG.error("==> {} events failed.", failedEvents.size()); } for (Iterator<ExecutionCommand> iterator = commandsToUpdate.iterator(); iterator.hasNext(); ) { ExecutionCommand cmd = iterator.next(); for (ServiceComponentHostEvent event : failedEvents.keySet()) { if (StringUtils.equals(event.getHostName(), cmd.getHostname()) && StringUtils.equals(event.getServiceComponentName(), cmd.getRole())) { iterator.remove(); commandsToAbort.put(cmd, failedEvents.get(event)); break; } } } } else { LOG.warn("There was events to process but cluster {} not found", stage.getClusterName()); } } LOG.debug("==> Scheduling {} tasks...", commandsToUpdate.size()); db.bulkHostRoleScheduled(stage, commandsToUpdate); if (commandsToAbort.size() > 0) { LOG.debug("==> Aborting {} tasks...", commandsToAbort.size()); List<Long> taskIds = new ArrayList<Long>(); for (ExecutionCommand command : commandsToAbort.keySet()) { taskIds.add(command.getTaskId()); } Collection<HostRoleCommand> hostRoleCommands = db.getTasks(taskIds); cancelHostRoleCommands(hostRoleCommands, FAILED_TASK_ABORT_REASONING); db.bulkAbortHostRole(stage, commandsToAbort); } LOG.debug("==> Adding {} tasks to queue...", commandsToUpdate.size()); for (ExecutionCommand cmd : commandsToUpdate) { if (Role.AMBARI_SERVER_ACTION.name().equals(cmd.getRole())) { serverActionExecutor.awake(); } else { commandsToEnqueue.put(cmd.getHostname(), cmd); } } actionQueue.enqueueAll(commandsToEnqueue.asMap()); LOG.debug("==> Finished."); if (!configuration.getParallelStageExecution()) { return; } if (exclusiveRequestIsGoing) { LOG.debug("Stage requires exclusive execution, skipping all executing any further stages"); break; } } requestsInProgress.retainAll(runningRequestIds); } finally { LOG.debug("Scheduler finished work."); unitOfWork.end(); } }
ActionScheduler implements Runnable { public void doWork() throws AmbariException { try { unitOfWork.begin(); threadEntityManager = entityManagerProvider.get(); processCancelledRequestsList(); if (db.getCommandsInProgressCount() == 0) { if (LOG.isDebugEnabled()) { LOG.debug("There are no stages currently in progress."); } actionQueue.updateListOfHostsWithPendingTask(null); return; } Set<Long> runningRequestIds = new HashSet<Long>(); List<Stage> stages = db.getStagesInProgress(); if (LOG.isDebugEnabled()) { LOG.debug("Scheduler wakes up"); LOG.debug("Processing {} in progress stages ", stages.size()); } publishInProgressTasks(stages); if (stages.isEmpty()) { if (LOG.isDebugEnabled()) { LOG.debug("There are no stages currently in progress."); } actionQueue.updateListOfHostsWithPendingTask(null); return; } int i_stage = 0; HashSet<String> hostsWithTasks = getListOfHostsWithPendingTask(stages); actionQueue.updateListOfHostsWithPendingTask(hostsWithTasks); stages = filterParallelPerHostStages(stages); boolean exclusiveRequestIsGoing = false; for (Stage stage : stages) { i_stage++; long requestId = stage.getRequestId(); LOG.debug("==> STAGE_i = " + i_stage + "(requestId=" + requestId + ",StageId=" + stage.getStageId() + ")"); RequestEntity request = db.getRequestEntity(requestId); if (request.isExclusive()) { if (runningRequestIds.size() > 0) { LOG.debug("Stage requires exclusive execution, but other requests are already executing. Stopping for now"); break; } exclusiveRequestIsGoing = true; } if (runningRequestIds.contains(requestId)) { LOG.debug("==> We don't want to process different stages from the same request in parallel"); continue; } else { runningRequestIds.add(requestId); if (!requestsInProgress.contains(requestId)) { requestsInProgress.add(requestId); db.startRequest(requestId); } } List<ExecutionCommand> commandsToSchedule = new ArrayList<>(); Multimap<String, AgentCommand> commandsToEnqueue = ArrayListMultimap.create(); Map<String, RoleStats> roleStats = processInProgressStage(stage, commandsToSchedule, commandsToEnqueue); boolean failed = false; for (Map.Entry<String, RoleStats> entry : roleStats.entrySet()) { String role = entry.getKey(); RoleStats stats = entry.getValue(); if (LOG.isDebugEnabled()) { LOG.debug("Stats for role: {}, stats={}", role, stats); } if (stats.isRoleFailed() && !stage.isSkippable()) { LOG.warn("{} failed, request {} will be aborted", role, request.getRequestId()); failed = true; break; } } if (!failed) { failed = hasPreviousStageFailed(stage); } if (failed) { LOG.error("Operation completely failed, aborting request id: {}", stage.getRequestId()); cancelHostRoleCommands(stage.getOrderedHostRoleCommands(), FAILED_TASK_ABORT_REASONING); abortOperationsForStage(stage); return; } List<ExecutionCommand> commandsToStart = new ArrayList<ExecutionCommand>(); List<ExecutionCommand> commandsToUpdate = new ArrayList<ExecutionCommand>(); for (ExecutionCommand cmd : commandsToSchedule) { ConfigHelper.processHiddenAttribute(cmd.getConfigurations(), cmd.getConfigurationAttributes(), cmd.getRole(), false); processHostRole(stage, cmd, commandsToStart, commandsToUpdate); } LOG.debug("==> Commands to start: {}", commandsToStart.size()); LOG.debug("==> Commands to update: {}", commandsToUpdate.size()); ListMultimap<String, ServiceComponentHostEvent> eventMap = formEventMap(stage, commandsToStart); Map<ExecutionCommand, String> commandsToAbort = new HashMap<ExecutionCommand, String>(); if (!eventMap.isEmpty()) { LOG.debug("==> processing {} serviceComponentHostEvents...", eventMap.size()); Cluster cluster = clusters.getCluster(stage.getClusterName()); if (cluster != null) { Map<ServiceComponentHostEvent, String> failedEvents = cluster.processServiceComponentHostEvents(eventMap); if (failedEvents.size() > 0) { LOG.error("==> {} events failed.", failedEvents.size()); } for (Iterator<ExecutionCommand> iterator = commandsToUpdate.iterator(); iterator.hasNext(); ) { ExecutionCommand cmd = iterator.next(); for (ServiceComponentHostEvent event : failedEvents.keySet()) { if (StringUtils.equals(event.getHostName(), cmd.getHostname()) && StringUtils.equals(event.getServiceComponentName(), cmd.getRole())) { iterator.remove(); commandsToAbort.put(cmd, failedEvents.get(event)); break; } } } } else { LOG.warn("There was events to process but cluster {} not found", stage.getClusterName()); } } LOG.debug("==> Scheduling {} tasks...", commandsToUpdate.size()); db.bulkHostRoleScheduled(stage, commandsToUpdate); if (commandsToAbort.size() > 0) { LOG.debug("==> Aborting {} tasks...", commandsToAbort.size()); List<Long> taskIds = new ArrayList<Long>(); for (ExecutionCommand command : commandsToAbort.keySet()) { taskIds.add(command.getTaskId()); } Collection<HostRoleCommand> hostRoleCommands = db.getTasks(taskIds); cancelHostRoleCommands(hostRoleCommands, FAILED_TASK_ABORT_REASONING); db.bulkAbortHostRole(stage, commandsToAbort); } LOG.debug("==> Adding {} tasks to queue...", commandsToUpdate.size()); for (ExecutionCommand cmd : commandsToUpdate) { if (Role.AMBARI_SERVER_ACTION.name().equals(cmd.getRole())) { serverActionExecutor.awake(); } else { commandsToEnqueue.put(cmd.getHostname(), cmd); } } actionQueue.enqueueAll(commandsToEnqueue.asMap()); LOG.debug("==> Finished."); if (!configuration.getParallelStageExecution()) { return; } if (exclusiveRequestIsGoing) { LOG.debug("Stage requires exclusive execution, skipping all executing any further stages"); break; } } requestsInProgress.retainAll(runningRequestIds); } finally { LOG.debug("Scheduler finished work."); unitOfWork.end(); } } }
ActionScheduler implements Runnable { public void doWork() throws AmbariException { try { unitOfWork.begin(); threadEntityManager = entityManagerProvider.get(); processCancelledRequestsList(); if (db.getCommandsInProgressCount() == 0) { if (LOG.isDebugEnabled()) { LOG.debug("There are no stages currently in progress."); } actionQueue.updateListOfHostsWithPendingTask(null); return; } Set<Long> runningRequestIds = new HashSet<Long>(); List<Stage> stages = db.getStagesInProgress(); if (LOG.isDebugEnabled()) { LOG.debug("Scheduler wakes up"); LOG.debug("Processing {} in progress stages ", stages.size()); } publishInProgressTasks(stages); if (stages.isEmpty()) { if (LOG.isDebugEnabled()) { LOG.debug("There are no stages currently in progress."); } actionQueue.updateListOfHostsWithPendingTask(null); return; } int i_stage = 0; HashSet<String> hostsWithTasks = getListOfHostsWithPendingTask(stages); actionQueue.updateListOfHostsWithPendingTask(hostsWithTasks); stages = filterParallelPerHostStages(stages); boolean exclusiveRequestIsGoing = false; for (Stage stage : stages) { i_stage++; long requestId = stage.getRequestId(); LOG.debug("==> STAGE_i = " + i_stage + "(requestId=" + requestId + ",StageId=" + stage.getStageId() + ")"); RequestEntity request = db.getRequestEntity(requestId); if (request.isExclusive()) { if (runningRequestIds.size() > 0) { LOG.debug("Stage requires exclusive execution, but other requests are already executing. Stopping for now"); break; } exclusiveRequestIsGoing = true; } if (runningRequestIds.contains(requestId)) { LOG.debug("==> We don't want to process different stages from the same request in parallel"); continue; } else { runningRequestIds.add(requestId); if (!requestsInProgress.contains(requestId)) { requestsInProgress.add(requestId); db.startRequest(requestId); } } List<ExecutionCommand> commandsToSchedule = new ArrayList<>(); Multimap<String, AgentCommand> commandsToEnqueue = ArrayListMultimap.create(); Map<String, RoleStats> roleStats = processInProgressStage(stage, commandsToSchedule, commandsToEnqueue); boolean failed = false; for (Map.Entry<String, RoleStats> entry : roleStats.entrySet()) { String role = entry.getKey(); RoleStats stats = entry.getValue(); if (LOG.isDebugEnabled()) { LOG.debug("Stats for role: {}, stats={}", role, stats); } if (stats.isRoleFailed() && !stage.isSkippable()) { LOG.warn("{} failed, request {} will be aborted", role, request.getRequestId()); failed = true; break; } } if (!failed) { failed = hasPreviousStageFailed(stage); } if (failed) { LOG.error("Operation completely failed, aborting request id: {}", stage.getRequestId()); cancelHostRoleCommands(stage.getOrderedHostRoleCommands(), FAILED_TASK_ABORT_REASONING); abortOperationsForStage(stage); return; } List<ExecutionCommand> commandsToStart = new ArrayList<ExecutionCommand>(); List<ExecutionCommand> commandsToUpdate = new ArrayList<ExecutionCommand>(); for (ExecutionCommand cmd : commandsToSchedule) { ConfigHelper.processHiddenAttribute(cmd.getConfigurations(), cmd.getConfigurationAttributes(), cmd.getRole(), false); processHostRole(stage, cmd, commandsToStart, commandsToUpdate); } LOG.debug("==> Commands to start: {}", commandsToStart.size()); LOG.debug("==> Commands to update: {}", commandsToUpdate.size()); ListMultimap<String, ServiceComponentHostEvent> eventMap = formEventMap(stage, commandsToStart); Map<ExecutionCommand, String> commandsToAbort = new HashMap<ExecutionCommand, String>(); if (!eventMap.isEmpty()) { LOG.debug("==> processing {} serviceComponentHostEvents...", eventMap.size()); Cluster cluster = clusters.getCluster(stage.getClusterName()); if (cluster != null) { Map<ServiceComponentHostEvent, String> failedEvents = cluster.processServiceComponentHostEvents(eventMap); if (failedEvents.size() > 0) { LOG.error("==> {} events failed.", failedEvents.size()); } for (Iterator<ExecutionCommand> iterator = commandsToUpdate.iterator(); iterator.hasNext(); ) { ExecutionCommand cmd = iterator.next(); for (ServiceComponentHostEvent event : failedEvents.keySet()) { if (StringUtils.equals(event.getHostName(), cmd.getHostname()) && StringUtils.equals(event.getServiceComponentName(), cmd.getRole())) { iterator.remove(); commandsToAbort.put(cmd, failedEvents.get(event)); break; } } } } else { LOG.warn("There was events to process but cluster {} not found", stage.getClusterName()); } } LOG.debug("==> Scheduling {} tasks...", commandsToUpdate.size()); db.bulkHostRoleScheduled(stage, commandsToUpdate); if (commandsToAbort.size() > 0) { LOG.debug("==> Aborting {} tasks...", commandsToAbort.size()); List<Long> taskIds = new ArrayList<Long>(); for (ExecutionCommand command : commandsToAbort.keySet()) { taskIds.add(command.getTaskId()); } Collection<HostRoleCommand> hostRoleCommands = db.getTasks(taskIds); cancelHostRoleCommands(hostRoleCommands, FAILED_TASK_ABORT_REASONING); db.bulkAbortHostRole(stage, commandsToAbort); } LOG.debug("==> Adding {} tasks to queue...", commandsToUpdate.size()); for (ExecutionCommand cmd : commandsToUpdate) { if (Role.AMBARI_SERVER_ACTION.name().equals(cmd.getRole())) { serverActionExecutor.awake(); } else { commandsToEnqueue.put(cmd.getHostname(), cmd); } } actionQueue.enqueueAll(commandsToEnqueue.asMap()); LOG.debug("==> Finished."); if (!configuration.getParallelStageExecution()) { return; } if (exclusiveRequestIsGoing) { LOG.debug("Stage requires exclusive execution, skipping all executing any further stages"); break; } } requestsInProgress.retainAll(runningRequestIds); } finally { LOG.debug("Scheduler finished work."); unitOfWork.end(); } } @Inject  ActionScheduler(@Named("schedulerSleeptime") long sleepTime,
                         @Named("actionTimeout") long actionTimeout, ActionDBAccessor db,
                         JPAEventPublisher jpaPublisher); protected  ActionScheduler(long sleepTimeMilliSec, long actionTimeoutMilliSec, ActionDBAccessor db,
                            ActionQueue actionQueue, Clusters fsmObject, int maxAttempts, HostsMap hostsMap,
                            UnitOfWork unitOfWork, AmbariEventPublisher ambariEventPublisher,
                            Configuration configuration, Provider<EntityManager> entityManagerProvider,
                            HostRoleCommandDAO hostRoleCommandDAO, HostRoleCommandFactory hostRoleCommandFactory); }
ActionScheduler implements Runnable { public void doWork() throws AmbariException { try { unitOfWork.begin(); threadEntityManager = entityManagerProvider.get(); processCancelledRequestsList(); if (db.getCommandsInProgressCount() == 0) { if (LOG.isDebugEnabled()) { LOG.debug("There are no stages currently in progress."); } actionQueue.updateListOfHostsWithPendingTask(null); return; } Set<Long> runningRequestIds = new HashSet<Long>(); List<Stage> stages = db.getStagesInProgress(); if (LOG.isDebugEnabled()) { LOG.debug("Scheduler wakes up"); LOG.debug("Processing {} in progress stages ", stages.size()); } publishInProgressTasks(stages); if (stages.isEmpty()) { if (LOG.isDebugEnabled()) { LOG.debug("There are no stages currently in progress."); } actionQueue.updateListOfHostsWithPendingTask(null); return; } int i_stage = 0; HashSet<String> hostsWithTasks = getListOfHostsWithPendingTask(stages); actionQueue.updateListOfHostsWithPendingTask(hostsWithTasks); stages = filterParallelPerHostStages(stages); boolean exclusiveRequestIsGoing = false; for (Stage stage : stages) { i_stage++; long requestId = stage.getRequestId(); LOG.debug("==> STAGE_i = " + i_stage + "(requestId=" + requestId + ",StageId=" + stage.getStageId() + ")"); RequestEntity request = db.getRequestEntity(requestId); if (request.isExclusive()) { if (runningRequestIds.size() > 0) { LOG.debug("Stage requires exclusive execution, but other requests are already executing. Stopping for now"); break; } exclusiveRequestIsGoing = true; } if (runningRequestIds.contains(requestId)) { LOG.debug("==> We don't want to process different stages from the same request in parallel"); continue; } else { runningRequestIds.add(requestId); if (!requestsInProgress.contains(requestId)) { requestsInProgress.add(requestId); db.startRequest(requestId); } } List<ExecutionCommand> commandsToSchedule = new ArrayList<>(); Multimap<String, AgentCommand> commandsToEnqueue = ArrayListMultimap.create(); Map<String, RoleStats> roleStats = processInProgressStage(stage, commandsToSchedule, commandsToEnqueue); boolean failed = false; for (Map.Entry<String, RoleStats> entry : roleStats.entrySet()) { String role = entry.getKey(); RoleStats stats = entry.getValue(); if (LOG.isDebugEnabled()) { LOG.debug("Stats for role: {}, stats={}", role, stats); } if (stats.isRoleFailed() && !stage.isSkippable()) { LOG.warn("{} failed, request {} will be aborted", role, request.getRequestId()); failed = true; break; } } if (!failed) { failed = hasPreviousStageFailed(stage); } if (failed) { LOG.error("Operation completely failed, aborting request id: {}", stage.getRequestId()); cancelHostRoleCommands(stage.getOrderedHostRoleCommands(), FAILED_TASK_ABORT_REASONING); abortOperationsForStage(stage); return; } List<ExecutionCommand> commandsToStart = new ArrayList<ExecutionCommand>(); List<ExecutionCommand> commandsToUpdate = new ArrayList<ExecutionCommand>(); for (ExecutionCommand cmd : commandsToSchedule) { ConfigHelper.processHiddenAttribute(cmd.getConfigurations(), cmd.getConfigurationAttributes(), cmd.getRole(), false); processHostRole(stage, cmd, commandsToStart, commandsToUpdate); } LOG.debug("==> Commands to start: {}", commandsToStart.size()); LOG.debug("==> Commands to update: {}", commandsToUpdate.size()); ListMultimap<String, ServiceComponentHostEvent> eventMap = formEventMap(stage, commandsToStart); Map<ExecutionCommand, String> commandsToAbort = new HashMap<ExecutionCommand, String>(); if (!eventMap.isEmpty()) { LOG.debug("==> processing {} serviceComponentHostEvents...", eventMap.size()); Cluster cluster = clusters.getCluster(stage.getClusterName()); if (cluster != null) { Map<ServiceComponentHostEvent, String> failedEvents = cluster.processServiceComponentHostEvents(eventMap); if (failedEvents.size() > 0) { LOG.error("==> {} events failed.", failedEvents.size()); } for (Iterator<ExecutionCommand> iterator = commandsToUpdate.iterator(); iterator.hasNext(); ) { ExecutionCommand cmd = iterator.next(); for (ServiceComponentHostEvent event : failedEvents.keySet()) { if (StringUtils.equals(event.getHostName(), cmd.getHostname()) && StringUtils.equals(event.getServiceComponentName(), cmd.getRole())) { iterator.remove(); commandsToAbort.put(cmd, failedEvents.get(event)); break; } } } } else { LOG.warn("There was events to process but cluster {} not found", stage.getClusterName()); } } LOG.debug("==> Scheduling {} tasks...", commandsToUpdate.size()); db.bulkHostRoleScheduled(stage, commandsToUpdate); if (commandsToAbort.size() > 0) { LOG.debug("==> Aborting {} tasks...", commandsToAbort.size()); List<Long> taskIds = new ArrayList<Long>(); for (ExecutionCommand command : commandsToAbort.keySet()) { taskIds.add(command.getTaskId()); } Collection<HostRoleCommand> hostRoleCommands = db.getTasks(taskIds); cancelHostRoleCommands(hostRoleCommands, FAILED_TASK_ABORT_REASONING); db.bulkAbortHostRole(stage, commandsToAbort); } LOG.debug("==> Adding {} tasks to queue...", commandsToUpdate.size()); for (ExecutionCommand cmd : commandsToUpdate) { if (Role.AMBARI_SERVER_ACTION.name().equals(cmd.getRole())) { serverActionExecutor.awake(); } else { commandsToEnqueue.put(cmd.getHostname(), cmd); } } actionQueue.enqueueAll(commandsToEnqueue.asMap()); LOG.debug("==> Finished."); if (!configuration.getParallelStageExecution()) { return; } if (exclusiveRequestIsGoing) { LOG.debug("Stage requires exclusive execution, skipping all executing any further stages"); break; } } requestsInProgress.retainAll(runningRequestIds); } finally { LOG.debug("Scheduler finished work."); unitOfWork.end(); } } @Inject  ActionScheduler(@Named("schedulerSleeptime") long sleepTime,
                         @Named("actionTimeout") long actionTimeout, ActionDBAccessor db,
                         JPAEventPublisher jpaPublisher); protected  ActionScheduler(long sleepTimeMilliSec, long actionTimeoutMilliSec, ActionDBAccessor db,
                            ActionQueue actionQueue, Clusters fsmObject, int maxAttempts, HostsMap hostsMap,
                            UnitOfWork unitOfWork, AmbariEventPublisher ambariEventPublisher,
                            Configuration configuration, Provider<EntityManager> entityManagerProvider,
                            HostRoleCommandDAO hostRoleCommandDAO, HostRoleCommandFactory hostRoleCommandFactory); void start(); void stop(); void awake(); @Override void run(); void doWork(); void scheduleCancellingRequest(long requestId, String reason); void setTaskTimeoutAdjustment(boolean val); @Subscribe void onEvent(EntityManagerCacheInvalidationEvent event); }
ActionScheduler implements Runnable { public void doWork() throws AmbariException { try { unitOfWork.begin(); threadEntityManager = entityManagerProvider.get(); processCancelledRequestsList(); if (db.getCommandsInProgressCount() == 0) { if (LOG.isDebugEnabled()) { LOG.debug("There are no stages currently in progress."); } actionQueue.updateListOfHostsWithPendingTask(null); return; } Set<Long> runningRequestIds = new HashSet<Long>(); List<Stage> stages = db.getStagesInProgress(); if (LOG.isDebugEnabled()) { LOG.debug("Scheduler wakes up"); LOG.debug("Processing {} in progress stages ", stages.size()); } publishInProgressTasks(stages); if (stages.isEmpty()) { if (LOG.isDebugEnabled()) { LOG.debug("There are no stages currently in progress."); } actionQueue.updateListOfHostsWithPendingTask(null); return; } int i_stage = 0; HashSet<String> hostsWithTasks = getListOfHostsWithPendingTask(stages); actionQueue.updateListOfHostsWithPendingTask(hostsWithTasks); stages = filterParallelPerHostStages(stages); boolean exclusiveRequestIsGoing = false; for (Stage stage : stages) { i_stage++; long requestId = stage.getRequestId(); LOG.debug("==> STAGE_i = " + i_stage + "(requestId=" + requestId + ",StageId=" + stage.getStageId() + ")"); RequestEntity request = db.getRequestEntity(requestId); if (request.isExclusive()) { if (runningRequestIds.size() > 0) { LOG.debug("Stage requires exclusive execution, but other requests are already executing. Stopping for now"); break; } exclusiveRequestIsGoing = true; } if (runningRequestIds.contains(requestId)) { LOG.debug("==> We don't want to process different stages from the same request in parallel"); continue; } else { runningRequestIds.add(requestId); if (!requestsInProgress.contains(requestId)) { requestsInProgress.add(requestId); db.startRequest(requestId); } } List<ExecutionCommand> commandsToSchedule = new ArrayList<>(); Multimap<String, AgentCommand> commandsToEnqueue = ArrayListMultimap.create(); Map<String, RoleStats> roleStats = processInProgressStage(stage, commandsToSchedule, commandsToEnqueue); boolean failed = false; for (Map.Entry<String, RoleStats> entry : roleStats.entrySet()) { String role = entry.getKey(); RoleStats stats = entry.getValue(); if (LOG.isDebugEnabled()) { LOG.debug("Stats for role: {}, stats={}", role, stats); } if (stats.isRoleFailed() && !stage.isSkippable()) { LOG.warn("{} failed, request {} will be aborted", role, request.getRequestId()); failed = true; break; } } if (!failed) { failed = hasPreviousStageFailed(stage); } if (failed) { LOG.error("Operation completely failed, aborting request id: {}", stage.getRequestId()); cancelHostRoleCommands(stage.getOrderedHostRoleCommands(), FAILED_TASK_ABORT_REASONING); abortOperationsForStage(stage); return; } List<ExecutionCommand> commandsToStart = new ArrayList<ExecutionCommand>(); List<ExecutionCommand> commandsToUpdate = new ArrayList<ExecutionCommand>(); for (ExecutionCommand cmd : commandsToSchedule) { ConfigHelper.processHiddenAttribute(cmd.getConfigurations(), cmd.getConfigurationAttributes(), cmd.getRole(), false); processHostRole(stage, cmd, commandsToStart, commandsToUpdate); } LOG.debug("==> Commands to start: {}", commandsToStart.size()); LOG.debug("==> Commands to update: {}", commandsToUpdate.size()); ListMultimap<String, ServiceComponentHostEvent> eventMap = formEventMap(stage, commandsToStart); Map<ExecutionCommand, String> commandsToAbort = new HashMap<ExecutionCommand, String>(); if (!eventMap.isEmpty()) { LOG.debug("==> processing {} serviceComponentHostEvents...", eventMap.size()); Cluster cluster = clusters.getCluster(stage.getClusterName()); if (cluster != null) { Map<ServiceComponentHostEvent, String> failedEvents = cluster.processServiceComponentHostEvents(eventMap); if (failedEvents.size() > 0) { LOG.error("==> {} events failed.", failedEvents.size()); } for (Iterator<ExecutionCommand> iterator = commandsToUpdate.iterator(); iterator.hasNext(); ) { ExecutionCommand cmd = iterator.next(); for (ServiceComponentHostEvent event : failedEvents.keySet()) { if (StringUtils.equals(event.getHostName(), cmd.getHostname()) && StringUtils.equals(event.getServiceComponentName(), cmd.getRole())) { iterator.remove(); commandsToAbort.put(cmd, failedEvents.get(event)); break; } } } } else { LOG.warn("There was events to process but cluster {} not found", stage.getClusterName()); } } LOG.debug("==> Scheduling {} tasks...", commandsToUpdate.size()); db.bulkHostRoleScheduled(stage, commandsToUpdate); if (commandsToAbort.size() > 0) { LOG.debug("==> Aborting {} tasks...", commandsToAbort.size()); List<Long> taskIds = new ArrayList<Long>(); for (ExecutionCommand command : commandsToAbort.keySet()) { taskIds.add(command.getTaskId()); } Collection<HostRoleCommand> hostRoleCommands = db.getTasks(taskIds); cancelHostRoleCommands(hostRoleCommands, FAILED_TASK_ABORT_REASONING); db.bulkAbortHostRole(stage, commandsToAbort); } LOG.debug("==> Adding {} tasks to queue...", commandsToUpdate.size()); for (ExecutionCommand cmd : commandsToUpdate) { if (Role.AMBARI_SERVER_ACTION.name().equals(cmd.getRole())) { serverActionExecutor.awake(); } else { commandsToEnqueue.put(cmd.getHostname(), cmd); } } actionQueue.enqueueAll(commandsToEnqueue.asMap()); LOG.debug("==> Finished."); if (!configuration.getParallelStageExecution()) { return; } if (exclusiveRequestIsGoing) { LOG.debug("Stage requires exclusive execution, skipping all executing any further stages"); break; } } requestsInProgress.retainAll(runningRequestIds); } finally { LOG.debug("Scheduler finished work."); unitOfWork.end(); } } @Inject  ActionScheduler(@Named("schedulerSleeptime") long sleepTime,
                         @Named("actionTimeout") long actionTimeout, ActionDBAccessor db,
                         JPAEventPublisher jpaPublisher); protected  ActionScheduler(long sleepTimeMilliSec, long actionTimeoutMilliSec, ActionDBAccessor db,
                            ActionQueue actionQueue, Clusters fsmObject, int maxAttempts, HostsMap hostsMap,
                            UnitOfWork unitOfWork, AmbariEventPublisher ambariEventPublisher,
                            Configuration configuration, Provider<EntityManager> entityManagerProvider,
                            HostRoleCommandDAO hostRoleCommandDAO, HostRoleCommandFactory hostRoleCommandFactory); void start(); void stop(); void awake(); @Override void run(); void doWork(); void scheduleCancellingRequest(long requestId, String reason); void setTaskTimeoutAdjustment(boolean val); @Subscribe void onEvent(EntityManagerCacheInvalidationEvent event); static final String FAILED_TASK_ABORT_REASONING; }