Repository: ambari Updated Branches: refs/heads/trunk 42c77f15a -> 57251f4b3
AMBARI-9897. RU: query for UpgradeItem is taking too long on large upgrades (ncole) Project: http://git-wip-us.apache.org/repos/asf/ambari/repo Commit: http://git-wip-us.apache.org/repos/asf/ambari/commit/57251f4b Tree: http://git-wip-us.apache.org/repos/asf/ambari/tree/57251f4b Diff: http://git-wip-us.apache.org/repos/asf/ambari/diff/57251f4b Branch: refs/heads/trunk Commit: 57251f4b3b048e3d38f1df0e769c6da70c25efbb Parents: 42c77f1 Author: Nate Cole <nc...@hortonworks.com> Authored: Tue Mar 3 10:17:43 2015 -0500 Committer: Nate Cole <nc...@hortonworks.com> Committed: Tue Mar 3 12:13:03 2015 -0500 ---------------------------------------------------------------------- .../controller/internal/CalculatedStatus.java | 70 +++++++++ .../internal/StageResourceProvider.java | 64 +++++---- .../internal/UpgradeGroupResourceProvider.java | 38 ++--- .../internal/UpgradeResourceProvider.java | 64 ++++++--- .../server/orm/dao/HostRoleCommandDAO.java | 56 ++++++++ .../dao/HostRoleCommandStatusSummaryDTO.java | 143 +++++++++++++++++++ .../ambari/server/state/UpgradeHelper.java | 35 ----- .../state/stack/upgrade/ColocatedGrouping.java | 9 +- .../internal/StageResourceProviderTest.java | 61 +++++--- .../internal/UpgradeResourceProviderTest.java | 51 +++++++ 10 files changed, 475 insertions(+), 116 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/ambari/blob/57251f4b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/CalculatedStatus.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/CalculatedStatus.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/CalculatedStatus.java index 3a84529..99c0a76 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/CalculatedStatus.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/CalculatedStatus.java @@ -17,15 +17,19 @@ */ package org.apache.ambari.server.controller.internal; +import java.util.ArrayList; import java.util.Collection; import java.util.HashMap; import java.util.HashSet; import java.util.LinkedList; import java.util.Map; +import java.util.Map.Entry; +import java.util.Set; import org.apache.ambari.server.actionmanager.HostRoleCommand; import org.apache.ambari.server.actionmanager.HostRoleStatus; import org.apache.ambari.server.actionmanager.Stage; +import org.apache.ambari.server.orm.dao.HostRoleCommandStatusSummaryDTO; import org.apache.ambari.server.orm.entities.HostRoleCommandEntity; import org.apache.ambari.server.orm.entities.StageEntity; @@ -228,6 +232,72 @@ public class CalculatedStatus { } /** + * Return counts of task statuses. + * @param stageDto the map of stage-to-summary value objects + * @param stageIds the stage ids to consider from the value objects + * @return the map of status to counts + */ + public static Map<HostRoleStatus, Integer> calculateTaskStatusCounts( + Map<Long, HostRoleCommandStatusSummaryDTO> stageDto, Set<Long> stageIds) { + + Map<HostRoleStatus, Integer> result = new HashMap<HostRoleStatus, Integer>(); + + for (Long stageId : stageIds) { + if (!stageDto.containsKey(stageId)) { + continue; + } + + HostRoleCommandStatusSummaryDTO dto = stageDto.get(stageId); + + for (Entry<HostRoleStatus, Integer> entry : dto.getCounts().entrySet()) { + if (!result.containsKey(entry.getKey())) { + result.put(entry.getKey(), 0); + } + + Integer old = result.get(entry.getKey()); + result.put(entry.getKey(), old + entry.getValue()); + } + + } + + return result; + } + + /** + * Calculates the overall status + * @param stageDto the map of stage-to-summary value objects + * @param stageIds the stage ids to consider from the value objects + * @return the calculated status + */ + public static CalculatedStatus statusFromStageSummary(Map<Long, HostRoleCommandStatusSummaryDTO> stageDto, + Set<Long> stageIds) { + + Collection<HostRoleStatus> stageStatuses = new HashSet<HostRoleStatus>(); + Collection<HostRoleStatus> taskStatuses = new ArrayList<HostRoleStatus>(); + + for (Long stageId : stageIds) { + if (!stageDto.containsKey(stageId)) { + continue; + } + + HostRoleCommandStatusSummaryDTO summary = stageDto.get(stageId); + HostRoleStatus stageStatus = calculateSummaryStatus(summary.getCounts(), + summary.getTaskTotal(), summary.isStageSkippable()); + + stageStatuses.add(stageStatus); + + taskStatuses.addAll(summary.getTaskStatuses()); + } + + // calculate the overall status from the stage statuses + HostRoleStatus status = calculateSummaryStatus(calculateStatusCounts(stageStatuses), stageStatuses.size(), false); + + double progressPercent = calculateProgressPercent(calculateStatusCounts(taskStatuses), taskStatuses.size()); + + return new CalculatedStatus(status, progressPercent); + } + + /** * Returns counts of tasks that are in various states. * * @param tasks the collection of tasks http://git-wip-us.apache.org/repos/asf/ambari/blob/57251f4b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/StageResourceProvider.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/StageResourceProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/StageResourceProvider.java index e55897c..fd6b751 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/StageResourceProvider.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/StageResourceProvider.java @@ -17,6 +17,20 @@ */ package org.apache.ambari.server.controller.internal; +import java.util.Collection; +import java.util.Collections; +import java.util.EnumSet; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Iterator; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import javax.inject.Inject; +import javax.inject.Provider; + import org.apache.ambari.server.StaticallyInject; import org.apache.ambari.server.actionmanager.HostRoleStatus; import org.apache.ambari.server.controller.AmbariManagementController; @@ -34,24 +48,13 @@ import org.apache.ambari.server.controller.spi.UnsupportedPropertyException; import org.apache.ambari.server.controller.utilities.PredicateBuilder; import org.apache.ambari.server.controller.utilities.PropertyHelper; import org.apache.ambari.server.orm.dao.HostRoleCommandDAO; +import org.apache.ambari.server.orm.dao.HostRoleCommandStatusSummaryDTO; import org.apache.ambari.server.orm.dao.StageDAO; import org.apache.ambari.server.orm.entities.HostRoleCommandEntity; import org.apache.ambari.server.orm.entities.StageEntity; import org.apache.ambari.server.state.Cluster; import org.apache.ambari.server.state.Clusters; -import javax.inject.Inject; -import javax.inject.Provider; -import java.util.Collection; -import java.util.EnumSet; -import java.util.HashMap; -import java.util.HashSet; -import java.util.Iterator; -import java.util.LinkedHashSet; -import java.util.List; -import java.util.Map; -import java.util.Set; - /** * ResourceProvider for Stage */ @@ -212,10 +215,18 @@ public class StageResourceProvider extends AbstractControllerResourceProvider im Set<Resource> results = new LinkedHashSet<Resource>(); Set<String> propertyIds = getRequestPropertyIds(request, predicate); + // !!! poor mans cache. toResource() shouldn't be calling the db + // every time, when the request id is likely the same for each stageEntity + Map<Long, Map<Long, HostRoleCommandStatusSummaryDTO>> cache = + new HashMap<Long, Map<Long, HostRoleCommandStatusSummaryDTO>>(); + List<StageEntity> entities = dao.findAll(request, predicate); for (StageEntity entity : entities) { - results.add(toResource(entity, propertyIds)); + results.add(toResource(cache, entity, propertyIds)); } + + cache.clear(); + return results; } @@ -301,8 +312,10 @@ public class StageResourceProvider extends AbstractControllerResourceProvider im * * @return the new resource */ - private Resource toResource(StageEntity entity, - Set<String> requestedIds) { + private Resource toResource( + Map<Long, Map<Long, HostRoleCommandStatusSummaryDTO>> cache, + StageEntity entity, + Set<String> requestedIds) { Resource resource = new ResourceImpl(Resource.Type.Stage); @@ -317,6 +330,12 @@ public class StageResourceProvider extends AbstractControllerResourceProvider im } } + if (!cache.containsKey(entity.getRequestId())) { + cache.put(entity.getRequestId(), hostRoleCommandDAO.findAggregateCounts(entity.getRequestId())); + } + + Map<Long, HostRoleCommandStatusSummaryDTO> summary = cache.get(entity.getRequestId()); + setResourceProperty(resource, STAGE_STAGE_ID, entity.getStageId(), requestedIds); setResourceProperty(resource, STAGE_REQUEST_ID, entity.getRequestId(), requestedIds); setResourceProperty(resource, STAGE_CONTEXT, entity.getRequestContext(), requestedIds); @@ -325,20 +344,17 @@ public class StageResourceProvider extends AbstractControllerResourceProvider im setResourceProperty(resource, STAGE_HOST_PARAMS, entity.getHostParamsStage(), requestedIds); setResourceProperty(resource, STAGE_SKIPPABLE, entity.isSkippable(), requestedIds); - Collection<HostRoleCommandEntity> tasks = entity.getHostRoleCommands(); - - Long startTime = tasks.isEmpty() ? 0L : Long.MAX_VALUE; - Long endTime = 0L; - - for (HostRoleCommandEntity task : tasks) { - startTime = Math.min(task.getStartTime(), startTime); - endTime = Math.max(task.getEndTime(), endTime); + Long startTime = Long.MAX_VALUE; + Long endTime = 0L; + if (summary.containsKey(entity.getStageId())) { + startTime = summary.get(entity.getStageId()).getStartTime(); + endTime = summary.get(entity.getStageId()).getEndTime(); } setResourceProperty(resource, STAGE_START_TIME, startTime, requestedIds); setResourceProperty(resource, STAGE_END_TIME, endTime, requestedIds); - CalculatedStatus status = CalculatedStatus.statusFromTaskEntities(tasks, entity.isSkippable()); + CalculatedStatus status = CalculatedStatus.statusFromStageSummary(summary, Collections.singleton(entity.getStageId())); setResourceProperty(resource, STAGE_PROGRESS_PERCENT, status.getPercent(), requestedIds); setResourceProperty(resource, STAGE_STATUS, status.getStatus().toString(), requestedIds); http://git-wip-us.apache.org/repos/asf/ambari/blob/57251f4b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/UpgradeGroupResourceProvider.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/UpgradeGroupResourceProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/UpgradeGroupResourceProvider.java index a0ff3d9..eb34d63 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/UpgradeGroupResourceProvider.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/UpgradeGroupResourceProvider.java @@ -17,7 +17,6 @@ */ package org.apache.ambari.server.controller.internal; -import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; import java.util.HashSet; @@ -38,10 +37,10 @@ import org.apache.ambari.server.controller.spi.Resource; import org.apache.ambari.server.controller.spi.ResourceAlreadyExistsException; import org.apache.ambari.server.controller.spi.SystemException; import org.apache.ambari.server.controller.spi.UnsupportedPropertyException; +import org.apache.ambari.server.orm.dao.HostRoleCommandDAO; +import org.apache.ambari.server.orm.dao.HostRoleCommandStatusSummaryDTO; import org.apache.ambari.server.orm.dao.StageDAO; import org.apache.ambari.server.orm.dao.UpgradeDAO; -import org.apache.ambari.server.orm.entities.HostRoleCommandEntity; -import org.apache.ambari.server.orm.entities.StageEntity; import org.apache.ambari.server.orm.entities.UpgradeEntity; import org.apache.ambari.server.orm.entities.UpgradeGroupEntity; import org.apache.ambari.server.orm.entities.UpgradeItemEntity; @@ -82,6 +81,9 @@ public class UpgradeGroupResourceProvider extends AbstractControllerResourceProv @Inject private static StageDAO stageDAO = null; + @Inject + private static HostRoleCommandDAO s_hostRoleCommandDao; + static { // properties PROPERTY_IDS.add(UPGRADE_REQUEST_ID); @@ -139,7 +141,9 @@ public class UpgradeGroupResourceProvider extends AbstractControllerResourceProv List<UpgradeGroupEntity> groups = upgrade.getUpgradeGroups(); if (null != groups) { - for (UpgradeGroupEntity group : upgrade.getUpgradeGroups()) { + Map<Long, HostRoleCommandStatusSummaryDTO> map = s_hostRoleCommandDao.findAggregateCounts(requestId); + + for (UpgradeGroupEntity group : groups) { Resource r = toResource(upgrade, group, requestPropertyIds); Set<Long> stageIds = new HashSet<Long>(); @@ -147,7 +151,7 @@ public class UpgradeGroupResourceProvider extends AbstractControllerResourceProv stageIds.add(itemEntity.getStageId()); } - aggregate(r, requestId, stageIds, requestPropertyIds); + aggregate(map, r, requestId, stageIds, requestPropertyIds); results.add(r); } @@ -196,15 +200,19 @@ public class UpgradeGroupResourceProvider extends AbstractControllerResourceProv * @param stageIds the set of resources ids of the stages * @param requestedIds the ids for the request */ - private void aggregate(Resource upgradeGroup, Long requestId, Set<Long> stageIds, Set<String> requestedIds) { - List<StageEntity> stages = stageDAO.findByStageIds(requestId, stageIds); + private void aggregate(Map<Long, HostRoleCommandStatusSummaryDTO> smap, + Resource upgradeGroup, Long requestId, Set<Long> stageIds, Set<String> requestedIds) { + + int size = 0; - List<HostRoleCommandEntity> tasks = new ArrayList<HostRoleCommandEntity>(); - for (StageEntity stage : stages) { - tasks.addAll(stage.getHostRoleCommands()); + Map<HostRoleStatus, Integer> counts = CalculatedStatus.calculateTaskStatusCounts(smap, stageIds); + + CalculatedStatus stageStatus = CalculatedStatus.statusFromStageSummary(smap, stageIds); + + for (Integer i : counts.values()) { + size += i.intValue(); } - Map<HostRoleStatus, Integer> counts = CalculatedStatus.calculateTaskEntityStatusCounts(tasks); Integer inProgress = 0; Integer completed = 0; @@ -215,13 +223,11 @@ public class UpgradeGroupResourceProvider extends AbstractControllerResourceProv inProgress += entry.getValue(); } } - setResourceProperty(upgradeGroup, UPGRADE_GROUP_TOTAL_TASKS, tasks.size(), requestedIds); + setResourceProperty(upgradeGroup, UPGRADE_GROUP_TOTAL_TASKS, size, requestedIds); setResourceProperty(upgradeGroup, UPGRADE_GROUP_IN_PROGRESS_TASKS, inProgress, requestedIds); setResourceProperty(upgradeGroup, UPGRADE_GROUP_COMPLETED_TASKS, completed, requestedIds); - CalculatedStatus status = CalculatedStatus.statusFromStageEntities(stages); - - setResourceProperty(upgradeGroup, UPGRADE_GROUP_STATUS, status.getStatus(), requestedIds); - setResourceProperty(upgradeGroup, UPGRADE_GROUP_PROGRESS_PERCENT, status.getPercent(), requestedIds); + setResourceProperty(upgradeGroup, UPGRADE_GROUP_STATUS, stageStatus.getStatus(), requestedIds); + setResourceProperty(upgradeGroup, UPGRADE_GROUP_PROGRESS_PERCENT, stageStatus.getPercent(), requestedIds); } } http://git-wip-us.apache.org/repos/asf/ambari/blob/57251f4b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/UpgradeResourceProvider.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/UpgradeResourceProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/UpgradeResourceProvider.java index 9a56e2b..4530726 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/UpgradeResourceProvider.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/UpgradeResourceProvider.java @@ -28,7 +28,6 @@ import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; -import java.util.Map.Entry; import java.util.Set; import java.util.regex.Matcher; import java.util.regex.Pattern; @@ -60,9 +59,13 @@ import org.apache.ambari.server.controller.spi.Resource; import org.apache.ambari.server.controller.spi.ResourceAlreadyExistsException; import org.apache.ambari.server.controller.spi.SystemException; import org.apache.ambari.server.controller.spi.UnsupportedPropertyException; +import org.apache.ambari.server.orm.dao.HostRoleCommandDAO; +import org.apache.ambari.server.orm.dao.HostRoleCommandStatusSummaryDTO; import org.apache.ambari.server.orm.dao.RepositoryVersionDAO; +import org.apache.ambari.server.orm.dao.RequestDAO; import org.apache.ambari.server.orm.dao.UpgradeDAO; import org.apache.ambari.server.orm.entities.RepositoryVersionEntity; +import org.apache.ambari.server.orm.entities.RequestEntity; import org.apache.ambari.server.orm.entities.UpgradeEntity; import org.apache.ambari.server.orm.entities.UpgradeGroupEntity; import org.apache.ambari.server.orm.entities.UpgradeItemEntity; @@ -109,6 +112,21 @@ public class UpgradeResourceProvider extends AbstractControllerResourceProvider protected static final String UPGRADE_REQUEST_STATUS = "Upgrade/request_status"; protected static final String UPGRADE_ABORT_REASON = "Upgrade/abort_reason"; + + /* + * Lifted from RequestResourceProvider + */ + private static final String REQUEST_CONTEXT_ID = "Upgrade/request_context"; + private static final String REQUEST_TYPE_ID = "Upgrade/type"; + private static final String REQUEST_CREATE_TIME_ID = "Upgrade/create_time"; + private static final String REQUEST_START_TIME_ID = "Upgrade/start_time"; + private static final String REQUEST_END_TIME_ID = "Upgrade/end_time"; + private static final String REQUEST_EXCLUSIVE_ID = "Upgrade/exclusive"; + + private static final String REQUEST_PROGRESS_PERCENT_ID = "Upgrade/progress_percent"; + private static final String REQUEST_STATUS_PROPERTY_ID = "Upgrade/request_status"; + + private static final Set<String> PK_PROPERTY_IDS = new HashSet<String>( Arrays.asList(UPGRADE_REQUEST_ID, UPGRADE_CLUSTER_NAME)); private static final Set<String> PROPERTY_IDS = new HashSet<String>(); @@ -145,6 +163,12 @@ public class UpgradeResourceProvider extends AbstractControllerResourceProvider @Inject private static Provider<AmbariCustomCommandExecutionHelper> s_commandExecutionHelper; + @Inject + private static RequestDAO s_requestDAO = null; + + @Inject + private static HostRoleCommandDAO s_hostRoleCommandDAO = null; + /** * Used to generated the correct tasks and stages during an upgrade. */ @@ -154,8 +178,6 @@ public class UpgradeResourceProvider extends AbstractControllerResourceProvider @Inject private static Configuration s_configuration; - private static Map<String, String> REQUEST_PROPERTY_MAP = new HashMap<String, String>(); - static { // properties PROPERTY_IDS.add(UPGRADE_CLUSTER_NAME); @@ -166,11 +188,14 @@ public class UpgradeResourceProvider extends AbstractControllerResourceProvider PROPERTY_IDS.add(UPGRADE_TO_VERSION); PROPERTY_IDS.add(UPGRADE_DIRECTION); - // !!! boo - for (String requestPropertyId : RequestResourceProvider.PROPERTY_IDS) { - REQUEST_PROPERTY_MAP.put(requestPropertyId, requestPropertyId.replace("Requests/", "Upgrade/")); - } - PROPERTY_IDS.addAll(REQUEST_PROPERTY_MAP.values()); + PROPERTY_IDS.add(REQUEST_CONTEXT_ID); + PROPERTY_IDS.add(REQUEST_CREATE_TIME_ID); + PROPERTY_IDS.add(REQUEST_END_TIME_ID); + PROPERTY_IDS.add(REQUEST_EXCLUSIVE_ID); + PROPERTY_IDS.add(REQUEST_PROGRESS_PERCENT_ID); + PROPERTY_IDS.add(REQUEST_START_TIME_ID); + PROPERTY_IDS.add(REQUEST_STATUS_PROPERTY_ID); + PROPERTY_IDS.add(REQUEST_TYPE_ID); // keys KEY_PROPERTY_IDS.put(Resource.Type.Upgrade, UPGRADE_REQUEST_ID); @@ -274,15 +299,22 @@ public class UpgradeResourceProvider extends AbstractControllerResourceProvider Resource r = toResource(entity, clusterName, requestPropertyIds); results.add(r); - // !!! not terribly efficient, but that's ok in this case. The handful-per-year - // an upgrade is done won't kill performance. - Resource r1 = s_upgradeHelper.getRequestResource(clusterName, - entity.getRequestId()); - for (Entry<String, String> entry : REQUEST_PROPERTY_MAP.entrySet()) { - Object o = r1.getPropertyValue(entry.getKey()); + RequestEntity rentity = s_requestDAO.findByPK(entity.getRequestId()); - setResourceProperty(r, entry.getValue(), o, requestPropertyIds); - } + setResourceProperty(r, REQUEST_CONTEXT_ID, rentity.getRequestContext(), requestPropertyIds); + setResourceProperty(r, REQUEST_TYPE_ID, rentity.getRequestType(), requestPropertyIds); + setResourceProperty(r, REQUEST_CREATE_TIME_ID, rentity.getCreateTime(), requestPropertyIds); + setResourceProperty(r, REQUEST_START_TIME_ID, rentity.getStartTime(), requestPropertyIds); + setResourceProperty(r, REQUEST_END_TIME_ID, rentity.getEndTime(), requestPropertyIds); + setResourceProperty(r, REQUEST_EXCLUSIVE_ID, rentity.isExclusive(), requestPropertyIds); + + Map<Long, HostRoleCommandStatusSummaryDTO> summary = s_hostRoleCommandDAO.findAggregateCounts(entity.getRequestId()); + + CalculatedStatus calc = CalculatedStatus.statusFromStageSummary( + summary, summary.keySet()); + + setResourceProperty(r, REQUEST_STATUS_PROPERTY_ID, calc.getStatus(), requestPropertyIds); + setResourceProperty(r, REQUEST_PROGRESS_PERCENT_ID, calc.getPercent(), requestPropertyIds); } } http://git-wip-us.apache.org/repos/asf/ambari/blob/57251f4b/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/HostRoleCommandDAO.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/HostRoleCommandDAO.java b/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/HostRoleCommandDAO.java index a55f383..c5d447a 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/HostRoleCommandDAO.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/HostRoleCommandDAO.java @@ -46,6 +46,26 @@ import com.google.inject.persist.Transactional; @Singleton public class HostRoleCommandDAO { + private static final String SUMMARY_DTO = String.format( + "SELECT NEW %s(" + + "MIN(hrc.startTime), " + + "MAX(hrc.endTime), " + + "MIN(hrc.stage.skippable), " + + "hrc.stageId, " + + "SUM(CASE WHEN hrc.status = :aborted THEN 1 ELSE 0 END), " + + "SUM(CASE WHEN hrc.status = :completed THEN 1 ELSE 0 END), " + + "SUM(CASE WHEN hrc.status = :failed THEN 1 ELSE 0 END), " + + "SUM(CASE WHEN hrc.status = :holding THEN 1 ELSE 0 END), " + + "SUM(CASE WHEN hrc.status = :holding_failed THEN 1 ELSE 0 END), " + + "SUM(CASE WHEN hrc.status = :holding_timedout THEN 1 ELSE 0 END), " + + "SUM(CASE WHEN hrc.status = :in_progress THEN 1 ELSE 0 END), " + + "SUM(CASE WHEN hrc.status = :pending THEN 1 ELSE 0 END), " + + "SUM(CASE WHEN hrc.status = :queued THEN 1 ELSE 0 END), " + + "SUM(CASE WHEN hrc.status = :timedout THEN 1 ELSE 0 END)" + + ") FROM HostRoleCommandEntity hrc " + + " GROUP BY hrc.requestId, hrc.stageId HAVING hrc.requestId = :requestId", + HostRoleCommandStatusSummaryDTO.class.getName()); + @Inject Provider<EntityManager> entityManagerProvider; @Inject @@ -349,4 +369,40 @@ public class HostRoleCommandDAO { public void removeByPK(int taskId) { remove(findByPK(taskId)); } + + + /** + * Finds the counts of tasks for a request and groups them by stage id. + * This allows for very efficient loading when there are a huge number of stages + * and tasks to iterate (for example, during a Rolling Upgrade). + * @param requestId the request id + * @return the map of stage-to-summary objects + */ + @RequiresSession + public Map<Long, HostRoleCommandStatusSummaryDTO> findAggregateCounts(Long requestId) { + + TypedQuery<HostRoleCommandStatusSummaryDTO> query = entityManagerProvider.get().createQuery( + SUMMARY_DTO, HostRoleCommandStatusSummaryDTO.class); + + query.setParameter("requestId", requestId); + query.setParameter("aborted", HostRoleStatus.ABORTED); + query.setParameter("completed", HostRoleStatus.COMPLETED); + query.setParameter("failed", HostRoleStatus.FAILED); + query.setParameter("holding", HostRoleStatus.HOLDING); + query.setParameter("holding_failed", HostRoleStatus.HOLDING_FAILED); + query.setParameter("holding_timedout", HostRoleStatus.HOLDING_TIMEDOUT); + query.setParameter("in_progress", HostRoleStatus.IN_PROGRESS); + query.setParameter("pending", HostRoleStatus.PENDING); + query.setParameter("queued", HostRoleStatus.QUEUED); + query.setParameter("timedout", HostRoleStatus.TIMEDOUT); + + Map<Long, HostRoleCommandStatusSummaryDTO> map = new HashMap<Long, HostRoleCommandStatusSummaryDTO>(); + + for (HostRoleCommandStatusSummaryDTO dto : daoUtils.selectList(query)) { + map.put(dto.getStageId(), dto); + } + + return map; + } + } http://git-wip-us.apache.org/repos/asf/ambari/blob/57251f4b/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/HostRoleCommandStatusSummaryDTO.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/HostRoleCommandStatusSummaryDTO.java b/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/HostRoleCommandStatusSummaryDTO.java new file mode 100644 index 0000000..ef867b0 --- /dev/null +++ b/ambari-server/src/main/java/org/apache/ambari/server/orm/dao/HostRoleCommandStatusSummaryDTO.java @@ -0,0 +1,143 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.ambari.server.orm.dao; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import org.apache.ambari.server.actionmanager.HostRoleStatus; + +/** + * Summary for a single stage with summary counts for the tasks for that stage. + */ +public class HostRoleCommandStatusSummaryDTO { + + private Long m_stageId = Long.valueOf(0L); + private Long m_minTime = Long.valueOf(0L); + private Long m_maxTime = Long.valueOf(Long.MAX_VALUE); + private boolean m_skippable = false; + private Map<HostRoleStatus, Integer> m_counts = new HashMap<HostRoleStatus, Integer>(); + private List<HostRoleStatus> m_tasksStatus = new ArrayList<HostRoleStatus>(); + + /** + * Constructor invoked by JPA. See {{@link HostRoleCommandDAO#findAggregateCounts(Long)}} + */ + public HostRoleCommandStatusSummaryDTO( + Number skippable, + Number minStartTime, + Number maxEndTime, + Number stageId, + Number aborted, + Number completed, + Number failed, + Number holding, + Number holdingFailed, + Number holdingTimedout, + Number inProgress, + Number pending, + Number queued, + Number timedout) { + + m_stageId = Long.valueOf(null == stageId ? 0L : stageId.longValue()); + if (null != skippable) { + m_skippable = (1 == skippable.intValue()); + } + if (null != minStartTime) { + m_minTime = Long.valueOf(minStartTime.longValue()); + } + if (null != maxEndTime) { + m_maxTime = Long.valueOf(maxEndTime.longValue()); + } + + put(HostRoleStatus.ABORTED, aborted); + put(HostRoleStatus.COMPLETED, completed); + put(HostRoleStatus.FAILED, failed); + put(HostRoleStatus.HOLDING, holding); + put(HostRoleStatus.HOLDING_FAILED, holdingFailed); + put(HostRoleStatus.HOLDING_TIMEDOUT, holdingTimedout); + put(HostRoleStatus.IN_PROGRESS, inProgress); + put(HostRoleStatus.PENDING, pending); + put(HostRoleStatus.QUEUED, queued); + put(HostRoleStatus.TIMEDOUT, timedout); + + } + + @SuppressWarnings("boxing") + private void put(HostRoleStatus status, Number number) { + if (null != number) { + m_counts.put(status, number.intValue()); + for (int i = 0; i < number.intValue(); i++) { + m_tasksStatus.add(status); + } + } else { + m_counts.put(status, 0); + } + } + + /** + * @return the stage id for this summary + */ + Long getStageId() { + return m_stageId; + } + + /** + * @return the task result counts, by status + */ + public Map<HostRoleStatus, Integer> getCounts() { + return m_counts; + } + + /** + * @return the list of tasks status, expanded to cover all tasks for the stage + */ + public List<HostRoleStatus> getTaskStatuses() { + return m_tasksStatus; + } + + /** + * @return the total number of tasks for the stage + */ + public int getTaskTotal() { + return m_tasksStatus.size(); + } + + /** + * @return {@code true} if the stage is skippable + */ + public boolean isStageSkippable() { + return m_skippable; + } + + /** + * @return the start time, minimum, for the tasks + */ + public Long getStartTime() { + return m_minTime; + } + + /** + * @return the end time, maximum, for the tasks + */ + public Long getEndTime() { + return m_maxTime; + } + +} http://git-wip-us.apache.org/repos/asf/ambari/blob/57251f4b/ambari-server/src/main/java/org/apache/ambari/server/state/UpgradeHelper.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/state/UpgradeHelper.java b/ambari-server/src/main/java/org/apache/ambari/server/state/UpgradeHelper.java index 76df049..0964335 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/state/UpgradeHelper.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/state/UpgradeHelper.java @@ -28,7 +28,6 @@ import java.util.regex.Pattern; import org.apache.ambari.server.AmbariException; import org.apache.ambari.server.api.services.AmbariMetaInfo; -import org.apache.ambari.server.controller.internal.RequestResourceProvider; import org.apache.ambari.server.controller.internal.StageResourceProvider; import org.apache.ambari.server.controller.predicate.AndPredicate; import org.apache.ambari.server.controller.spi.ClusterController; @@ -483,40 +482,6 @@ public class UpgradeHelper { } /** - * Gets a Request resource to aggreate with an Upgrade - * @param clusterName the cluster name - * @param requestId the request id - * @return the resource for the Request - * @throws UnsupportedPropertyException - * @throws NoSuchResourceException - * @throws NoSuchParentResourceException - * @throws SystemException - */ - // !!! FIXME this feels very wrong - public Resource getRequestResource(String clusterName, Long requestId) - throws UnsupportedPropertyException, NoSuchResourceException, NoSuchParentResourceException, SystemException { - - ClusterController clusterController = ClusterControllerHelper.getClusterController(); - - Request request = PropertyHelper.getReadRequest(); - - Predicate predicate = new PredicateBuilder().property(RequestResourceProvider.REQUEST_CLUSTER_NAME_PROPERTY_ID).equals(clusterName).and() - // !!! RequestResourceProvider is expecting a string, not a Long for the requestId - .property(RequestResourceProvider.REQUEST_ID_PROPERTY_ID).equals(requestId.toString()).toPredicate(); - - QueryResponse response = clusterController.getResources(Resource.Type.Request, - request, predicate); - - Set<Resource> resources = response.getResources(); - if (1 != resources.size()) { - throw new SystemException(String.format( - "Cannot uniquely identify the request resource for %s", requestId)); - } - - return resources.iterator().next(); - } - - /** * Helper to set service and component display names on the context * @param context the context to update * @param service the service name http://git-wip-us.apache.org/repos/asf/ambari/blob/57251f4b/ambari-server/src/main/java/org/apache/ambari/server/state/stack/upgrade/ColocatedGrouping.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/state/stack/upgrade/ColocatedGrouping.java b/ambari-server/src/main/java/org/apache/ambari/server/state/stack/upgrade/ColocatedGrouping.java index 6252d4e..a8e9c43 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/state/stack/upgrade/ColocatedGrouping.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/state/stack/upgrade/ColocatedGrouping.java @@ -22,6 +22,7 @@ import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.LinkedHashMap; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Map.Entry; @@ -236,7 +237,7 @@ public class ColocatedGrouping extends Grouping { * @param wrappers the list of stage wrappers */ private void formatFirstBatch(UpgradeContext ctx, ManualTask task, List<StageWrapper> wrappers) { - List<String> compNames = new ArrayList<String>(); + Set<String> names = new LinkedHashSet<String>(); Map<String, Set<String>> compLocations = new HashMap<String, Set<String>>(); for (StageWrapper sw : wrappers) { @@ -251,7 +252,7 @@ public class ColocatedGrouping extends Grouping { compLocations.get(host).add(tw.getComponent()); } - compNames.add(ctx.getComponentDisplay( + names.add(ctx.getComponentDisplay( tw.getService(), tw.getComponent())); } } @@ -261,9 +262,11 @@ public class ColocatedGrouping extends Grouping { if (task.message.contains("{{components}}")) { StringBuilder sb = new StringBuilder(); + List<String> compNames = new ArrayList<String>(names); + if (compNames.size() == 1) { sb.append(compNames.get(0)); - } else if (compNames.size() > 1) { + } else if (names.size() > 1) { String last = compNames.remove(compNames.size() - 1); sb.append(StringUtils.join(compNames, ", ")); sb.append(" and ").append(last); http://git-wip-us.apache.org/repos/asf/ambari/blob/57251f4b/ambari-server/src/test/java/org/apache/ambari/server/controller/internal/StageResourceProviderTest.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/test/java/org/apache/ambari/server/controller/internal/StageResourceProviderTest.java b/ambari-server/src/test/java/org/apache/ambari/server/controller/internal/StageResourceProviderTest.java index 1d80f47..96a92ad 100644 --- a/ambari-server/src/test/java/org/apache/ambari/server/controller/internal/StageResourceProviderTest.java +++ b/ambari-server/src/test/java/org/apache/ambari/server/controller/internal/StageResourceProviderTest.java @@ -17,11 +17,25 @@ */ package org.apache.ambari.server.controller.internal; -import com.google.inject.Binder; -import com.google.inject.Guice; -import com.google.inject.Injector; -import com.google.inject.Module; -import com.google.inject.util.Modules; +import static org.easymock.EasyMock.anyLong; +import static org.easymock.EasyMock.anyObject; +import static org.easymock.EasyMock.createNiceMock; +import static org.easymock.EasyMock.createStrictMock; +import static org.easymock.EasyMock.eq; +import static org.easymock.EasyMock.expect; +import static org.easymock.EasyMock.replay; +import static org.easymock.EasyMock.verify; +import static org.junit.Assert.fail; + +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.Set; + import org.apache.ambari.server.actionmanager.ActionManager; import org.apache.ambari.server.actionmanager.HostRoleStatus; import org.apache.ambari.server.controller.AmbariManagementController; @@ -33,32 +47,23 @@ import org.apache.ambari.server.controller.utilities.PredicateBuilder; import org.apache.ambari.server.controller.utilities.PropertyHelper; import org.apache.ambari.server.metadata.ActionMetadata; import org.apache.ambari.server.orm.InMemoryDefaultTestModule; +import org.apache.ambari.server.orm.dao.HostRoleCommandDAO; +import org.apache.ambari.server.orm.dao.HostRoleCommandStatusSummaryDTO; import org.apache.ambari.server.orm.dao.StageDAO; import org.apache.ambari.server.orm.entities.HostRoleCommandEntity; import org.apache.ambari.server.orm.entities.StageEntity; import org.apache.ambari.server.state.Cluster; import org.apache.ambari.server.state.Clusters; +import org.easymock.EasyMock; import org.junit.Assert; import org.junit.Before; import org.junit.Test; -import java.util.Collection; -import java.util.Collections; -import java.util.HashSet; -import java.util.LinkedList; -import java.util.List; -import java.util.Map; -import java.util.Set; - -import static org.easymock.EasyMock.anyLong; -import static org.easymock.EasyMock.anyObject; -import static org.easymock.EasyMock.createNiceMock; -import static org.easymock.EasyMock.createStrictMock; -import static org.easymock.EasyMock.eq; -import static org.easymock.EasyMock.expect; -import static org.easymock.EasyMock.replay; -import static org.easymock.EasyMock.verify; -import static org.junit.Assert.*; +import com.google.inject.Binder; +import com.google.inject.Guice; +import com.google.inject.Injector; +import com.google.inject.Module; +import com.google.inject.util.Modules; public class StageResourceProviderTest { @@ -67,12 +72,23 @@ public class StageResourceProviderTest { private Cluster cluster = null; private AmbariManagementController managementController = null; private Injector injector; + private HostRoleCommandDAO hrcDao = null; @Before public void before() { dao = createStrictMock(StageDAO.class); clusters = createStrictMock(Clusters.class); cluster = createStrictMock(Cluster.class); + hrcDao = createStrictMock(HostRoleCommandDAO.class); + + expect(hrcDao.findAggregateCounts(EasyMock.anyObject(Long.class))).andReturn( + new HashMap<Long, HostRoleCommandStatusSummaryDTO>() {{ + put(0L, new HostRoleCommandStatusSummaryDTO( + 0, 1000L, 2500L, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0)); + }}).anyTimes(); + + replay(hrcDao); + managementController = createNiceMock(AmbariManagementController.class); // create an injector which will inject the mocks @@ -243,6 +259,7 @@ public class StageResourceProviderTest { binder.bind(StageDAO.class).toInstance(dao); binder.bind(Clusters.class).toInstance(clusters); binder.bind(Cluster.class).toInstance(cluster); + binder.bind(HostRoleCommandDAO.class).toInstance(hrcDao); binder.bind(AmbariManagementController.class).toInstance(managementController); binder.bind(ActionMetadata.class); } http://git-wip-us.apache.org/repos/asf/ambari/blob/57251f4b/ambari-server/src/test/java/org/apache/ambari/server/controller/internal/UpgradeResourceProviderTest.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/test/java/org/apache/ambari/server/controller/internal/UpgradeResourceProviderTest.java b/ambari-server/src/test/java/org/apache/ambari/server/controller/internal/UpgradeResourceProviderTest.java index ef4bd82..baa885c 100644 --- a/ambari-server/src/test/java/org/apache/ambari/server/controller/internal/UpgradeResourceProviderTest.java +++ b/ambari-server/src/test/java/org/apache/ambari/server/controller/internal/UpgradeResourceProviderTest.java @@ -32,6 +32,7 @@ import java.util.Set; import org.apache.ambari.server.actionmanager.ActionManager; import org.apache.ambari.server.actionmanager.HostRoleCommand; +import org.apache.ambari.server.actionmanager.HostRoleStatus; import org.apache.ambari.server.actionmanager.RequestStatus; import org.apache.ambari.server.actionmanager.Stage; import org.apache.ambari.server.api.resources.UpgradeResourceDefinition; @@ -46,9 +47,11 @@ import org.apache.ambari.server.controller.utilities.PropertyHelper; import org.apache.ambari.server.orm.GuiceJpaInitializer; import org.apache.ambari.server.orm.InMemoryDefaultTestModule; import org.apache.ambari.server.orm.OrmTestHelper; +import org.apache.ambari.server.orm.dao.HostRoleCommandDAO; import org.apache.ambari.server.orm.dao.RepositoryVersionDAO; import org.apache.ambari.server.orm.dao.StageDAO; import org.apache.ambari.server.orm.dao.UpgradeDAO; +import org.apache.ambari.server.orm.entities.HostRoleCommandEntity; import org.apache.ambari.server.orm.entities.RepositoryVersionEntity; import org.apache.ambari.server.orm.entities.StageEntity; import org.apache.ambari.server.orm.entities.UpgradeEntity; @@ -539,6 +542,54 @@ public class UpgradeResourceProviderTest { } + @Test + public void testPercents() throws Exception { + org.apache.ambari.server.controller.spi.RequestStatus status = testCreateResources(); + + Set<Resource> createdResources = status.getAssociatedResources(); + assertEquals(1, createdResources.size()); + Resource res = createdResources.iterator().next(); + Long id = (Long) res.getPropertyValue("Upgrade/request_id"); + assertNotNull(id); + assertEquals(Long.valueOf(1), id); + + StageDAO stageDao = injector.getInstance(StageDAO.class); + HostRoleCommandDAO hrcDao = injector.getInstance(HostRoleCommandDAO.class); + + List<StageEntity> stages = stageDao.findByRequestId(id); + List<HostRoleCommandEntity> tasks = hrcDao.findByRequest(id); + + Set<Long> stageIds = new HashSet<Long>(); + for (StageEntity se : stages) { + stageIds.add(se.getStageId()); + } + + CalculatedStatus calc = null; + int i = 0; + for (HostRoleCommandEntity hrce : tasks) { + hrce.setStatus(HostRoleStatus.IN_PROGRESS); + hrcDao.merge(hrce); + calc = CalculatedStatus.statusFromStageSummary(hrcDao.findAggregateCounts(id), stageIds); + assertEquals(((i++) + 1) * 4.375d, calc.getPercent(), 0.01d); + assertEquals(HostRoleStatus.IN_PROGRESS, calc.getStatus()); + } + + i = 0; + for (HostRoleCommandEntity hrce : tasks) { + hrce.setStatus(HostRoleStatus.COMPLETED); + hrcDao.merge(hrce); + calc = CalculatedStatus.statusFromStageSummary(hrcDao.findAggregateCounts(id), stageIds); + assertEquals(35 + (((i++) + 1) * 8.125), calc.getPercent(), 0.01d); + if (i < 8) { + assertEquals(HostRoleStatus.IN_PROGRESS, calc.getStatus()); + } + } + + calc = CalculatedStatus.statusFromStageSummary(hrcDao.findAggregateCounts(id), stageIds); + assertEquals(HostRoleStatus.COMPLETED, calc.getStatus()); + assertEquals(100d, calc.getPercent(), 0.01d); + } + /** * @param amc * @return the provider