@Test public void testServerActionInMultipleRequests() 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 clusterName = "cluster1"; String hostname1 = "ahost.ambari.apache.org"; String hostname2 = "bhost.ambari.apache.org"; HashMap<String, ServiceComponentHost> hosts = new HashMap<String, ServiceComponentHost>(); hosts.put(hostname1, sch); hosts.put(hostname2, sch); hosts.put(Stage.INTERNAL_HOSTNAME, sch); when(scomp.getServiceComponentHosts()).thenReturn(hosts); List<Stage> stages = new ArrayList<Stage>(); Stage stage01 = createStage(clusterName, 0, 1); addTask(stage01, Stage.INTERNAL_HOSTNAME, clusterName, Role.AMBARI_SERVER_ACTION, RoleCommand.ACTIONEXECUTE, "AMBARI", 1); Stage stage11 = createStage("cluster1", 1, 1); addTask(stage11, hostname1, clusterName, Role.KERBEROS_CLIENT, RoleCommand.CUSTOM_COMMAND, "KERBEROS", 2); Stage stage02 = createStage("cluster1", 0, 2); addTask(stage02, Stage.INTERNAL_HOSTNAME, clusterName, Role.AMBARI_SERVER_ACTION, RoleCommand.ACTIONEXECUTE, "AMBARI", 3); Stage stage12 = createStage("cluster1", 1, 2); addTask(stage12, hostname2, clusterName, Role.KERBEROS_CLIENT, RoleCommand.CUSTOM_COMMAND, "KERBEROS", 4); stages.add(stage01); stages.add(stage11); stages.add(stage02); stages.add(stage12); 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); Properties properties = new Properties(); properties.put(Configuration.PARALLEL_STAGE_EXECUTION.getKey(), "true"); Configuration conf = new Configuration(properties); ActionScheduler scheduler = new ActionScheduler(100, 50, db, aq, fsm, 3, new HostsMap((String) null), unitOfWork, EasyMock.createNiceMock(AmbariEventPublisher.class), conf, entityManagerProviderMock, hostRoleCommandDAOMock, (HostRoleCommandFactory)null); scheduler.doWork(); Assert.assertEquals(HostRoleStatus.QUEUED, stages.get(0).getHostRoleStatus(Stage.INTERNAL_HOSTNAME, Role.AMBARI_SERVER_ACTION.name())); Assert.assertEquals(HostRoleStatus.PENDING, stages.get(1).getHostRoleStatus(hostname1, Role.KERBEROS_CLIENT.name())); Assert.assertEquals(HostRoleStatus.QUEUED, stages.get(2).getHostRoleStatus(Stage.INTERNAL_HOSTNAME, Role.AMBARI_SERVER_ACTION.name())); Assert.assertEquals(HostRoleStatus.PENDING, stages.get(3).getHostRoleStatus(hostname2, Role.KERBEROS_CLIENT.name())); }
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; }