/*
 * 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.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.List;
import java.util.Map;
import java.util.Set;

import org.apache.ambari.server.Role;
import org.apache.ambari.server.actionmanager.HostRoleCommand;
import org.apache.ambari.server.actionmanager.HostRoleStatus;
import org.apache.ambari.server.actionmanager.Request;
import org.apache.ambari.server.actionmanager.Stage;
import org.apache.ambari.server.events.listeners.tasks.TaskStatusListener;
import org.apache.ambari.server.orm.dao.HostRoleCommandDAO;
import org.apache.ambari.server.orm.dao.HostRoleCommandStatusSummaryDTO;
import org.apache.ambari.server.orm.entities.HostRoleCommandEntity;
import org.apache.ambari.server.orm.entities.StageEntity;
import org.apache.ambari.server.orm.entities.StageEntityPK;
import org.apache.ambari.server.topology.LogicalRequest;
import org.apache.ambari.server.topology.TopologyManager;

import com.google.common.base.Function;
import com.google.common.base.Predicate;
import com.google.common.collect.FluentIterable;

/**
 * Status of a request resource, calculated from a set of tasks or stages.
 */
public class CalculatedStatus {

  /**
   * The calculated overall status.
   */
  private final HostRoleStatus status;

  /**
   * The display status.
   */
  private final HostRoleStatus displayStatus;

  /**
   * The calculated percent complete.
   */
  private final double percent;

  /**
   * A status which represents a COMPLETED state at 100%
   */
  public static final CalculatedStatus COMPLETED = new CalculatedStatus(HostRoleStatus.COMPLETED,
      HostRoleStatus.COMPLETED, 100.0);

  /**
   * A status which represents a PENDING state at 0%
   */
  public static final CalculatedStatus PENDING = new CalculatedStatus(HostRoleStatus.PENDING,
      HostRoleStatus.PENDING, 0.0);

  /**
   * A status which represents an ABORTED state at -1%
   */
  public static final CalculatedStatus ABORTED = new CalculatedStatus(HostRoleStatus.ABORTED, HostRoleStatus.ABORTED, -1);

  // ----- Constructors ------------------------------------------------------

  /**
   * Constructor.
   *
   * @param status   the calculated overall status
   * @param percent  the calculated percent complete
   */
  private CalculatedStatus(HostRoleStatus status, double percent) {
    this(status, null, percent);
  }

  /**
   * Overloaded constructor that allows to set the display status if required.
   *
   * @param status   the calculated overall status
   * @param displayStatus the calculated display status
   * @param percent  the calculated percent complete
   */
  private CalculatedStatus(HostRoleStatus status, HostRoleStatus displayStatus, double percent) {
    this.status  = status;
    this.displayStatus  = displayStatus;
    this.percent = percent;
  }


  // ----- CalculatedStatus --------------------------------------------------

  /**
   * Get the calculated status.
   *
   * @return the status
   */
  public HostRoleStatus getStatus() {
    return status;
  }

  /**
   * Get the calculated display status. The display_status field is used as
   * a hint for UI. It's effective only on UpgradeGroup level.
   * We should expose it for the following states:
   * SKIPPED_FAILED
   * FAILED
   *
   * @return the display status
   */
  public HostRoleStatus getDisplayStatus() {
    return displayStatus;
  }

  /**
   * Get the calculated percent complete.
   *
   * @return the percent complete
   */
  public double getPercent() {
    return percent;
  }


  // ----- helper methods ----------------------------------------------------

  /**
   * Factory method to create a calculated status.  Calculate request status from the given
   * collection of task entities.
   *
   * @param tasks      the collection of task entities
   * @param skippable  true if a single failed status should NOT result in an overall failed status
   *
   * @return a calculated status
   */
  public static CalculatedStatus statusFromTaskEntities(Collection<HostRoleCommandEntity> tasks, boolean skippable) {

    int size = tasks.size();

    Map<HostRoleStatus, Integer> taskStatusCounts = CalculatedStatus.calculateTaskEntityStatusCounts(tasks);

    HostRoleStatus status = calculateSummaryStatus(taskStatusCounts, size, skippable);

    double progressPercent = calculateProgressPercent(taskStatusCounts, size);

    return new CalculatedStatus(status, progressPercent);
  }

  /**
   * Factory method to create a calculated status.  Calculate request status from the given
   * collection of stage entities.
   *
   * @param stages  the collection of stage entities
   *
   * @return a calculated status
   */
  public static CalculatedStatus statusFromStageEntities(Collection<StageEntity> stages) {
    Collection<HostRoleStatus> stageStatuses = new HashSet<>();
    Collection<HostRoleCommandEntity> tasks = new HashSet<>();

    for (StageEntity stage : stages) {
      // get all the tasks for the stage
      Collection<HostRoleCommandEntity> stageTasks = stage.getHostRoleCommands();

      // calculate the stage status from the task status counts
      HostRoleStatus stageStatus =
          calculateSummaryStatus(calculateTaskEntityStatusCounts(stageTasks), stageTasks.size(), stage.isSkippable());

      stageStatuses.add(stageStatus);

      // keep track of all of the tasks for all stages
      tasks.addAll(stageTasks);
    }

    // calculate the overall status from the stage statuses
    HostRoleStatus status = calculateSummaryStatusOfUpgrade(calculateStatusCounts(stageStatuses), stageStatuses.size());

    // calculate the progress from the task status counts
    double progressPercent = calculateProgressPercent(calculateTaskEntityStatusCounts(tasks), tasks.size());

    return new CalculatedStatus(status, progressPercent);
  }

  /**
   * Factory method to create a calculated status.  Calculate request status from the given
   * collection of stages.
   *
   * @param stages  the collection of stages
   *
   * @return a calculated status
   */
  public static CalculatedStatus statusFromStages(Collection<Stage> stages) {

    Collection<HostRoleStatus> stageStatuses = new HashSet<>();
    Collection<HostRoleCommand> tasks = new HashSet<>();

    for (Stage stage : stages) {
      // get all the tasks for the stage
      Collection<HostRoleCommand> stageTasks = stage.getOrderedHostRoleCommands();

      // calculate the stage status from the task status counts
      HostRoleStatus stageStatus =
          calculateSummaryStatus(calculateTaskStatusCounts(stageTasks), stageTasks.size(), stage.isSkippable());

      stageStatuses.add(stageStatus);

      // keep track of all of the tasks for all stages
      tasks.addAll(stageTasks);
    }

    // calculate the overall status from the stage statuses
    HostRoleStatus status = calculateSummaryStatusOfUpgrade(calculateStatusCounts(stageStatuses), stageStatuses.size());

    // calculate the progress from the task status counts
    double progressPercent = calculateProgressPercent(calculateTaskStatusCounts(tasks), tasks.size());

    return new CalculatedStatus(status, progressPercent);
  }

  /**
   * Returns counts of tasks that are in various states.
   *
   * @param hostRoleStatuses  the collection of tasks
   *
   * @return a map of counts of tasks keyed by the task status
   */
  public static Map<HostRoleStatus, Integer> calculateStatusCounts(Collection<HostRoleStatus> hostRoleStatuses) {
    Map<HostRoleStatus, Integer> counters = new HashMap<>();
    // initialize
    for (HostRoleStatus hostRoleStatus : HostRoleStatus.values()) {
      counters.put(hostRoleStatus, 0);
    }
    // calculate counts
    for (HostRoleStatus status : hostRoleStatuses) {
      // count tasks where isCompletedState() == true as COMPLETED
      // but don't count tasks with COMPLETED status twice
      if (status.isCompletedState() && status != HostRoleStatus.COMPLETED) {
        // Increase total number of completed tasks;
        counters.put(HostRoleStatus.COMPLETED, counters.get(HostRoleStatus.COMPLETED) + 1);
      }
      // Increment counter for particular status
      counters.put(status, counters.get(status) + 1);
    }

    // We overwrite the value to have the sum converged
    counters.put(HostRoleStatus.IN_PROGRESS,
        hostRoleStatuses.size() -
            counters.get(HostRoleStatus.COMPLETED) -
            counters.get(HostRoleStatus.QUEUED) -
            counters.get(HostRoleStatus.PENDING));

    return counters;
  }

  /**
   * Returns counts of tasks that are in various states.
   *
   * @param hostRoleCommands  collection of beans {@link HostRoleCommand}
   *
   * @return a map of counts of tasks keyed by the task status
   */
  public static Map<HostRoleStatus, Integer> calculateStatusCountsForTasks(Collection<HostRoleCommand> hostRoleCommands) {
    Map<HostRoleStatus, Integer> counters = new HashMap<>();
    // initialize
    for (HostRoleStatus hostRoleStatus : HostRoleStatus.values()) {
      counters.put(hostRoleStatus, 0);
    }
    // calculate counts
    for (HostRoleCommand hrc : hostRoleCommands) {
      // count tasks where isCompletedState() == true as COMPLETED
      // but don't count tasks with COMPLETED status twice
      if (hrc.getStatus().isCompletedState() && hrc.getStatus() != HostRoleStatus.COMPLETED) {
        // Increase total number of completed tasks;
        counters.put(HostRoleStatus.COMPLETED, counters.get(HostRoleStatus.COMPLETED) + 1);
      }
      // Increment counter for particular status
      counters.put(hrc.getStatus(), counters.get(hrc.getStatus()) + 1);
    }

    // We overwrite the value to have the sum converged
    counters.put(HostRoleStatus.IN_PROGRESS,
        hostRoleCommands.size() -
            counters.get(HostRoleStatus.COMPLETED) -
            counters.get(HostRoleStatus.QUEUED) -
            counters.get(HostRoleStatus.PENDING));

    return counters;
  }

  /**
   * Returns map for counts of stages that are in various states.
   *
   * @param stages  collection of beans {@link org.apache.ambari.server.events.listeners.tasks.TaskStatusListener.ActiveStage}
   *
   * @return a map of counts of tasks keyed by the task status
   */
  public static Map<StatusType,Map<HostRoleStatus, Integer>> calculateStatusCountsForStage(Collection<TaskStatusListener.ActiveStage> stages) {

    Map<StatusType,Map<HostRoleStatus, Integer>> counters = new HashMap<>();
    for (StatusType statusType : StatusType.values()) {
      Map <HostRoleStatus, Integer> statusMap = new HashMap<>();
      counters.put(statusType,statusMap);
      // initialize
      for (HostRoleStatus hostRoleStatus : HostRoleStatus.values()) {
        statusMap.put(hostRoleStatus, 0);
      }
      for (TaskStatusListener.ActiveStage stage : stages) {
        // count tasks where isCompletedState() == true as COMPLETED
        // but don't count tasks with COMPLETED status twice
        HostRoleStatus status;
        if (statusType == StatusType.DISPLAY_STATUS) {
          status = stage.getDisplayStatus();
        } else {
          status = stage.getStatus();
        }
        if (status.isCompletedState() && status != HostRoleStatus.COMPLETED) {
          // Increase total number of completed tasks;
          statusMap.put(HostRoleStatus.COMPLETED, statusMap.get(HostRoleStatus.COMPLETED) + 1);
        }

        // Increment counter for particular status
        statusMap.put(status, statusMap.get(status) + 1);
      }
      statusMap.put(HostRoleStatus.IN_PROGRESS,
          stages.size() -
              statusMap.get(HostRoleStatus.COMPLETED) -
              statusMap.get(HostRoleStatus.QUEUED) -
              statusMap.get(HostRoleStatus.PENDING));
    }
    return counters;
  }


  /**
   * Returns counts of tasks that are in various states.
   *
   * @param hostRoleCommands  collection of beans {@link HostRoleCommand}
   *
   * @return a map of counts of tasks keyed by the task status
   */
  public static Map<HostRoleStatus, Integer> calculateStatusCountsForTasks(Collection<HostRoleCommand> hostRoleCommands, StageEntityPK stage) {
    Map<HostRoleStatus, Integer> counters = new HashMap<>();
    List<HostRoleCommand> hostRoleCommandsOfStage = new ArrayList<>();
    // initialize
    for (HostRoleStatus hostRoleStatus : HostRoleStatus.values()) {
      counters.put(hostRoleStatus, 0);
    }
    // calculate counts
    for (HostRoleCommand hrc : hostRoleCommands) {
      if (stage.getStageId() == hrc.getStageId() && stage.getRequestId() == hrc.getRequestId()) {
        // count tasks where isCompletedState() == true as COMPLETED
        // but don't count tasks with COMPLETED status twice
        if (hrc.getStatus().isCompletedState() && hrc.getStatus() != HostRoleStatus.COMPLETED) {
          // Increase total number of completed tasks;
          counters.put(HostRoleStatus.COMPLETED, counters.get(HostRoleStatus.COMPLETED) + 1);
        }

        // Increment counter for particular status
        counters.put(hrc.getStatus(), counters.get(hrc.getStatus()) + 1);

        hostRoleCommandsOfStage.add(hrc);
      }
    }

    // We overwrite the value to have the sum converged
    counters.put(HostRoleStatus.IN_PROGRESS,
        hostRoleCommandsOfStage.size() -
            counters.get(HostRoleStatus.COMPLETED) -
            counters.get(HostRoleStatus.QUEUED) -
            counters.get(HostRoleStatus.PENDING));

    return counters;
  }

  /**
   * Returns counts of task entities that are in various states.
   *
   * @param tasks  the collection of task entities
   *
   * @return a map of counts of tasks keyed by the task status
   */
  public static Map<HostRoleStatus, Integer> calculateTaskEntityStatusCounts(Collection<HostRoleCommandEntity> tasks) {
    Collection<HostRoleStatus> hostRoleStatuses = new LinkedList<>();

    for (HostRoleCommandEntity hostRoleCommand : tasks) {
      hostRoleStatuses.add(hostRoleCommand.getStatus());
    }
    return calculateStatusCounts(hostRoleStatuses);
  }

  /**
   * 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) {

    List<HostRoleStatus> status = new ArrayList<>();

    for (Long stageId : stageIds) {
      if (!stageDto.containsKey(stageId)) {
        continue;
      }

      HostRoleCommandStatusSummaryDTO dto = stageDto.get(stageId);

      status.addAll(dto.getTaskStatuses());
    }

    return calculateStatusCounts(status);
  }

  /**
   * Calculates the status for specified request by id.
   * @param s_hostRoleCommandDAO is used to retrieve the map of stage-to-summary value objects
   * @param topologyManager topology manager
   * @param requestId the request id
   * @return the calculated status
   */
  public static CalculatedStatus statusFromRequest(HostRoleCommandDAO s_hostRoleCommandDAO,
                                                   TopologyManager topologyManager, Long requestId) {
    Map<Long, HostRoleCommandStatusSummaryDTO> summary = s_hostRoleCommandDAO.findAggregateCounts(requestId);

    // get summaries from TopologyManager for logical requests
    summary.putAll(topologyManager.getStageSummaries(requestId));

    // summary might be empty due to delete host have cleared all
    // HostRoleCommands or due to hosts haven't registered yet with the cluster
    // when the cluster is provisioned with a Blueprint
    final CalculatedStatus status;
    LogicalRequest logicalRequest = topologyManager.getRequest(requestId);
    if (summary.isEmpty() && null != logicalRequest) {
      // In this case, it appears that there are no tasks but this is a logical
      // topology request, so it's a matter of hosts simply not registering yet
      // for tasks to be created ==> status = PENDING.
      // For a new LogicalRequest there should be at least one HostRequest,
      // while if they were removed already ==> status = COMPLETED.
      if (logicalRequest.getHostRequests().isEmpty()) {
        status = CalculatedStatus.COMPLETED;
      } else {
        status = CalculatedStatus.PENDING;
      }
    } else {
      // there are either tasks or this is not a logical request, so do normal
      // status calculations
      status = CalculatedStatus.statusFromStageSummary(summary, summary.keySet());
    }
    return status;
  }

  /**
   * Calculates the overall status of an upgrade. If there are no tasks, then a
   * status of {@link HostRoleStatus#COMPLETED} is returned.
   *
   * @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) {

    // if either are empty, then we have no tasks and therefore no status - we
    // should return COMPLETED. This can happen if someone removes all tasks but
    // leaves the stages and request
    if (stageDto.isEmpty() || stageIds.isEmpty()) {
      return COMPLETED;
    }

    Collection<HostRoleStatus> stageStatuses = new HashSet<>();
    Collection<HostRoleStatus> stageDisplayStatuses = new HashSet<>();
    Collection<HostRoleStatus> taskStatuses = new ArrayList<>();

    for (Long stageId : stageIds) {
      if (!stageDto.containsKey(stageId)) {
        continue;
      }

      HostRoleCommandStatusSummaryDTO summary = stageDto.get(stageId);

      int total = summary.getTaskTotal();
      boolean skip = summary.isStageSkippable();
      Map<HostRoleStatus, Integer> counts = calculateStatusCounts(summary.getTaskStatuses());
      HostRoleStatus stageStatus = calculateSummaryStatus(counts, total, skip);
      HostRoleStatus stageDisplayStatus = calculateSummaryDisplayStatus(counts, total, skip);

      stageStatuses.add(stageStatus);
      stageDisplayStatuses.add(stageDisplayStatus);
      taskStatuses.addAll(summary.getTaskStatuses());
    }

    // calculate the overall status from the stage statuses
    Map<HostRoleStatus, Integer> counts = calculateStatusCounts(stageStatuses);
    Map<HostRoleStatus, Integer> displayCounts = calculateStatusCounts(stageDisplayStatuses);

    HostRoleStatus status = calculateSummaryStatusOfUpgrade(counts, stageStatuses.size());
    HostRoleStatus displayStatus = calculateSummaryDisplayStatus(displayCounts, stageDisplayStatuses.size(), false);

    double progressPercent = calculateProgressPercent(calculateStatusCounts(taskStatuses), taskStatuses.size());

    return new CalculatedStatus(status, displayStatus, progressPercent);
  }

  /**
   * Returns counts of tasks that are in various states.
   *
   * @param tasks  the collection of tasks
   *
   * @return a map of counts of tasks keyed by the task status
   */
  private static Map<HostRoleStatus, Integer> calculateTaskStatusCounts(Collection<HostRoleCommand> tasks) {
    Collection<HostRoleStatus> hostRoleStatuses = new LinkedList<>();

    for (HostRoleCommand hostRoleCommand : tasks) {
      hostRoleStatuses.add(hostRoleCommand.getStatus());
    }
    return calculateStatusCounts(hostRoleStatuses);
  }

  /**
   * Calculate the percent complete based on the given status counts.
   *
   * @param counters  counts of resources that are in various states
   * @param total     total number of resources in request
   *
   * @return the percent complete for the stage
   */
  private static double calculateProgressPercent(Map<HostRoleStatus, Integer> counters, double total) {
    return total == 0 ? 0 :
        ((counters.get(HostRoleStatus.QUEUED)              * 0.09 +
          counters.get(HostRoleStatus.IN_PROGRESS)         * 0.35 +
          counters.get(HostRoleStatus.HOLDING)             * 0.35 +
          counters.get(HostRoleStatus.HOLDING_FAILED)      * 0.35 +
          counters.get(HostRoleStatus.HOLDING_TIMEDOUT)    * 0.35 +
          counters.get(HostRoleStatus.COMPLETED)) / total) * 100.0;
  }

  /**
   * Calculate overall status of a stage or upgrade based on the given status counts.
   *
   * @param counters   counts of resources that are in various states
   * @param total      total number of resources in request. NOTE, completed tasks may be counted twice.
   * @param skippable  true if a single failed status should NOT result in an overall failed status return
   *
   * @return summary request status based on statuses of tasks in different states.
   */
  public static HostRoleStatus calculateSummaryStatus(Map<HostRoleStatus, Integer> counters,
      int total, boolean skippable) {

    // when there are 0 tasks, return COMPLETED
    if (total == 0) {
      return HostRoleStatus.COMPLETED;
    }

    if (counters.get(HostRoleStatus.PENDING) == total) {
      return HostRoleStatus.PENDING;
    }

    // By definition, any tasks in a future stage must be held in a PENDING status.
    if (counters.get(HostRoleStatus.HOLDING) > 0 || counters.get(HostRoleStatus.HOLDING_FAILED) > 0 || counters.get(HostRoleStatus.HOLDING_TIMEDOUT) > 0) {
      return counters.get(HostRoleStatus.HOLDING) > 0 ? HostRoleStatus.HOLDING :
      counters.get(HostRoleStatus.HOLDING_FAILED) > 0 ? HostRoleStatus.HOLDING_FAILED :
      HostRoleStatus.HOLDING_TIMEDOUT;
    }

    // Because tasks are not skippable, guaranteed to be FAILED
    if (counters.get(HostRoleStatus.FAILED) > 0 && !skippable) {
      return HostRoleStatus.FAILED;
    }

    // Because tasks are not skippable, guaranteed to be TIMEDOUT
    if (counters.get(HostRoleStatus.TIMEDOUT) > 0  && !skippable) {
      return HostRoleStatus.TIMEDOUT;
    }

    int numActiveTasks = counters.get(HostRoleStatus.PENDING) + counters.get(HostRoleStatus.QUEUED) + counters.get(HostRoleStatus.IN_PROGRESS);
    // ABORTED only if there are no other active tasks
    if (counters.get(HostRoleStatus.ABORTED) > 0 && numActiveTasks == 0) {
      return HostRoleStatus.ABORTED;
    }

    if (counters.get(HostRoleStatus.COMPLETED) == total) {
      return HostRoleStatus.COMPLETED;
    }

    return HostRoleStatus.IN_PROGRESS;
  }

  /**
   *
   * @param counters counts of resources that are in various states
   * @param skippable {Boolean} <code>TRUE<code/> if failure of any of the task should not fail the stage
   * @return {@link HostRoleStatus}
   */
  public static HostRoleStatus calculateSummaryStatusFromPartialSet(Map<HostRoleStatus, Integer> counters,
                                                      boolean skippable) {

    HostRoleStatus status = HostRoleStatus.PENDING;
    // By definition, any tasks in a future stage must be held in a PENDING status.
    if (counters.get(HostRoleStatus.HOLDING) > 0 || counters.get(HostRoleStatus.HOLDING_FAILED) > 0 || counters.get(HostRoleStatus.HOLDING_TIMEDOUT) > 0) {
      status =  counters.get(HostRoleStatus.HOLDING) > 0 ? HostRoleStatus.HOLDING :
          counters.get(HostRoleStatus.HOLDING_FAILED) > 0 ? HostRoleStatus.HOLDING_FAILED :
              HostRoleStatus.HOLDING_TIMEDOUT;
    }

    // Because tasks are not skippable, guaranteed to be FAILED
    if (counters.get(HostRoleStatus.FAILED) > 0 && !skippable) {
      status = HostRoleStatus.FAILED;
    }

    // Because tasks are not skippable, guaranteed to be TIMEDOUT
    if (counters.get(HostRoleStatus.TIMEDOUT) > 0  && !skippable) {
      status = HostRoleStatus.TIMEDOUT;
    }

    int inProgressTasks =  counters.get(HostRoleStatus.QUEUED) + counters.get(HostRoleStatus.IN_PROGRESS);
    if (inProgressTasks > 0) {
      status = HostRoleStatus.IN_PROGRESS;
    }

    return status;
  }


  /**
   *
   * @param hostRoleCommands list of {@link HostRoleCommand} for a stage
   * @param counters counts of resources that are in various states
   * @param successFactors Map of roles to their successfactor for a stage
   * @param skippable {Boolean} <code>TRUE<code/> if failure of any of the task should not fail the stage
   * @return {@link HostRoleStatus} based on success factor
   */
  public static HostRoleStatus calculateStageStatus(List <HostRoleCommand> hostRoleCommands, Map<HostRoleStatus, Integer> counters, Map<Role, Float> successFactors,
                                                    boolean skippable) {

    // when there are 0 tasks, return COMPLETED
    int total = hostRoleCommands.size();
    if (total == 0) {
      return HostRoleStatus.COMPLETED;
    }

    if (counters.get(HostRoleStatus.PENDING) == total) {
      return HostRoleStatus.PENDING;
    }

    // By definition, any tasks in a future stage must be held in a PENDING status.
    if (counters.get(HostRoleStatus.HOLDING) > 0 || counters.get(HostRoleStatus.HOLDING_FAILED) > 0 || counters.get(HostRoleStatus.HOLDING_TIMEDOUT) > 0) {
      return counters.get(HostRoleStatus.HOLDING) > 0 ? HostRoleStatus.HOLDING :
          counters.get(HostRoleStatus.HOLDING_FAILED) > 0 ? HostRoleStatus.HOLDING_FAILED :
              HostRoleStatus.HOLDING_TIMEDOUT;
    }


    if (counters.get(HostRoleStatus.FAILED) > 0 && !skippable) {
      Set<Role> rolesWithFailedTasks = getRolesOfFailedTasks(hostRoleCommands);
      Boolean didStageFailed = didStageFailed(hostRoleCommands, rolesWithFailedTasks, successFactors);
      if (didStageFailed) return HostRoleStatus.FAILED;
    }


    if (counters.get(HostRoleStatus.TIMEDOUT) > 0  && !skippable) {
      Set<Role> rolesWithTimedOutTasks = getRolesOfTimedOutTasks(hostRoleCommands);
      Boolean didStageFailed = didStageFailed(hostRoleCommands, rolesWithTimedOutTasks, successFactors);
      if (didStageFailed) return HostRoleStatus.TIMEDOUT;
    }

    int numActiveTasks = counters.get(HostRoleStatus.PENDING) + counters.get(HostRoleStatus.QUEUED) + counters.get(HostRoleStatus.IN_PROGRESS);

    if (numActiveTasks > 0) {
      return HostRoleStatus.IN_PROGRESS;
    } else if (counters.get(HostRoleStatus.ABORTED) > 0) {
      Set<Role> rolesWithTimedOutTasks = getRolesOfAbortedTasks(hostRoleCommands);
      Boolean didStageFailed = didStageFailed(hostRoleCommands, rolesWithTimedOutTasks, successFactors);
      if (didStageFailed) return HostRoleStatus.ABORTED;
    }

    return HostRoleStatus.COMPLETED;
  }

  /**
   *  Get all {@link Role} any of whose tasks is in {@link HostRoleStatus#FAILED}
   * @param hostRoleCommands list of {@link HostRoleCommand}
   * @return Set of {@link Role}
   */
  protected static Set<Role> getRolesOfFailedTasks(List <HostRoleCommand> hostRoleCommands) {
    return getRolesOfTasks(hostRoleCommands, HostRoleStatus.FAILED);
  }

  /**
   *  Get all {@link Role} any of whose tasks is in {@link HostRoleStatus#TIMEDOUT}
   * @param hostRoleCommands list of {@link HostRoleCommand}
   * @return Set of {@link Role}
   */
  protected static Set<Role> getRolesOfTimedOutTasks(List <HostRoleCommand> hostRoleCommands) {
    return getRolesOfTasks(hostRoleCommands, HostRoleStatus.TIMEDOUT);
  }

  /**
   *  Get all {@link Role} any of whose tasks is in {@link HostRoleStatus#ABORTED}
   * @param hostRoleCommands list of {@link HostRoleCommand}
   * @return Set of {@link Role}
   */
  protected static Set<Role> getRolesOfAbortedTasks(List <HostRoleCommand> hostRoleCommands) {
    return getRolesOfTasks(hostRoleCommands, HostRoleStatus.ABORTED);
  }

  /**
   * Get all {@link Role} any of whose tasks are in given {@code status}
   * @param hostRoleCommands list of {@link HostRoleCommand}
   * @param status {@link HostRoleStatus}
   * @return Set of {@link Role}
   */
  protected static Set<Role> getRolesOfTasks(List <HostRoleCommand> hostRoleCommands, final HostRoleStatus status) {

    Predicate<HostRoleCommand> predicate = new Predicate<HostRoleCommand>() {
      @Override
      public boolean apply(HostRoleCommand hrc) {
        return hrc.getStatus() ==  status;
      }
    };

    Function<HostRoleCommand, Role> transform = new Function<HostRoleCommand, Role>() {
      @Override
      public Role apply(HostRoleCommand hrc) {
        return hrc.getRole();
      }
    };
    return FluentIterable.from(hostRoleCommands)
        .filter(predicate)
        .transform(transform)
        .toSet();
  }

  /**
   *
   * @param hostRoleCommands list of {@link HostRoleCommand} for a stage
   * @param roles  set of roles to be checked for meeting success criteria
   * @param successFactors  map of role to it's success factor
   * @return {Boolean} <code>TRUE</code> if stage failed due to hostRoleCommands of any role not meeting success criteria
   */
  protected static Boolean didStageFailed(List<HostRoleCommand> hostRoleCommands, Set<Role> roles, Map<Role, Float> successFactors) {
    Boolean isFailed = Boolean.FALSE;
    for (Role role: roles) {
      List <HostRoleCommand> hostRoleCommandsOfRole = getHostRoleCommandsOfRole(hostRoleCommands, role);
      List <HostRoleCommand> failedHostRoleCommands =  getFailedHostRoleCommands(hostRoleCommandsOfRole);
      float successRatioForRole = (hostRoleCommandsOfRole.size() - failedHostRoleCommands.size())/hostRoleCommandsOfRole.size();
      Float successFactorForRole =  successFactors.get(role) == null ? 1.0f : successFactors.get(role);
      if (successRatioForRole  < successFactorForRole) {
        isFailed = Boolean.TRUE;
        break;
      }
    }
    return isFailed;
  }

  /**
   *
   * @param hostRoleCommands list of {@link HostRoleCommand}
   * @param role {@link Role}
   * @return list of {@link HostRoleCommand} that belongs to {@link Role}
   */
  protected static List<HostRoleCommand> getHostRoleCommandsOfRole(List <HostRoleCommand> hostRoleCommands, final Role role) {
    Predicate<HostRoleCommand> predicate = new Predicate<HostRoleCommand>() {
      @Override
      public boolean apply(HostRoleCommand hrc) {
        return hrc.getRole() ==  role;
      }
    };
    return FluentIterable.from(hostRoleCommands)
        .filter(predicate)
        .toList();
  }

  /**
   *
   * @param hostRoleCommands list of {@link HostRoleCommand}
   * @return list of {@link HostRoleCommand} with failed status
   */
  protected static List<HostRoleCommand> getFailedHostRoleCommands(List <HostRoleCommand> hostRoleCommands) {
    Predicate<HostRoleCommand> predicate = new Predicate<HostRoleCommand>() {
      @Override
      public boolean apply(HostRoleCommand hrc) {
        return hrc.getStatus().isFailedAndNotSkippableState();
      }
    };
    return FluentIterable.from(hostRoleCommands)
        .filter(predicate)
        .toList();
  }


  /**
   * Calculate overall status from collection of statuses
   * @param hostRoleStatuses list of all stage's {@link HostRoleStatus}
   * @return overall status of a request
   */
  public static HostRoleStatus getOverallStatusForRequest (Collection<HostRoleStatus> hostRoleStatuses) {
    Map<HostRoleStatus, Integer> statusCount = calculateStatusCounts(hostRoleStatuses);
    return calculateSummaryStatus(statusCount, hostRoleStatuses.size(), false);
  }

  /**
   * Calculate overall display status from collection of statuses
   * @param hostRoleStatuses list of all stage's {@link HostRoleStatus}
   * @return overall display status of a request
   */
  public static HostRoleStatus getOverallDisplayStatusForRequest (Collection<HostRoleStatus> hostRoleStatuses) {
    Map<HostRoleStatus, Integer> statusCount = calculateStatusCounts(hostRoleStatuses);
    return calculateSummaryDisplayStatus(statusCount, hostRoleStatuses.size(), false);
  }


  /**
   * Calculate overall status of an upgrade.
   *
   * @param counters   counts of resources that are in various states
   * @param total      total number of resources in request
   *
