/*
 * 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.text.MessageFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

import org.apache.ambari.annotations.Experimental;
import org.apache.ambari.annotations.ExperimentalFeature;
import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.Role;
import org.apache.ambari.server.RoleCommand;
import org.apache.ambari.server.StaticallyInject;
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.RequestFactory;
import org.apache.ambari.server.actionmanager.Stage;
import org.apache.ambari.server.actionmanager.StageFactory;
import org.apache.ambari.server.agent.ExecutionCommand.KeyNames;
import org.apache.ambari.server.api.services.AmbariMetaInfo;
import org.apache.ambari.server.cleanup.ClasspathScannerUtils;
import org.apache.ambari.server.configuration.Configuration;
import org.apache.ambari.server.controller.ActionExecutionContext;
import org.apache.ambari.server.controller.AmbariActionExecutionHelper;
import org.apache.ambari.server.controller.AmbariCustomCommandExecutionHelper;
import org.apache.ambari.server.controller.AmbariManagementController;
import org.apache.ambari.server.controller.ExecuteCommandJson;
import org.apache.ambari.server.controller.KerberosHelper;
import org.apache.ambari.server.controller.KerberosHelperImpl.SupportedCustomOperation;
import org.apache.ambari.server.controller.UpdateConfigurationPolicy;
import org.apache.ambari.server.controller.spi.NoSuchParentResourceException;
import org.apache.ambari.server.controller.spi.NoSuchResourceException;
import org.apache.ambari.server.controller.spi.Predicate;
import org.apache.ambari.server.controller.spi.Request;
import org.apache.ambari.server.controller.spi.RequestStatus;
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.events.UpdateEventType;
import org.apache.ambari.server.events.UpgradeUpdateEvent;
import org.apache.ambari.server.events.publishers.STOMPUpdatePublisher;
import org.apache.ambari.server.orm.dao.HostRoleCommandDAO;
import org.apache.ambari.server.orm.dao.HostRoleCommandStatusSummaryDTO;
import org.apache.ambari.server.orm.dao.RequestDAO;
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.RequestEntity;
import org.apache.ambari.server.orm.entities.StackEntity;
import org.apache.ambari.server.orm.entities.UpgradeEntity;
import org.apache.ambari.server.orm.entities.UpgradeGroupEntity;
import org.apache.ambari.server.orm.entities.UpgradeHistoryEntity;
import org.apache.ambari.server.orm.entities.UpgradeItemEntity;
import org.apache.ambari.server.security.authorization.AuthorizationException;
import org.apache.ambari.server.security.authorization.AuthorizationHelper;
import org.apache.ambari.server.security.authorization.ResourceType;
import org.apache.ambari.server.security.authorization.RoleAuthorization;
import org.apache.ambari.server.serveraction.ServerAction;
import org.apache.ambari.server.serveraction.kerberos.KerberosOperationException;
import org.apache.ambari.server.serveraction.upgrades.PluginUpgradeServerAction;
import org.apache.ambari.server.stack.upgrade.AddComponentTask;
import org.apache.ambari.server.stack.upgrade.ConfigUpgradePack;
import org.apache.ambari.server.stack.upgrade.ConfigureTask;
import org.apache.ambari.server.stack.upgrade.CreateAndConfigureTask;
import org.apache.ambari.server.stack.upgrade.Direction;
import org.apache.ambari.server.stack.upgrade.ManualTask;
import org.apache.ambari.server.stack.upgrade.ServerSideActionTask;
import org.apache.ambari.server.stack.upgrade.Task;
import org.apache.ambari.server.stack.upgrade.UpdateStackGrouping;
import org.apache.ambari.server.stack.upgrade.UpgradePack;
import org.apache.ambari.server.stack.upgrade.UpgradeScope;
import org.apache.ambari.server.stack.upgrade.orchestrate.StageWrapper;
import org.apache.ambari.server.stack.upgrade.orchestrate.TaskWrapper;
import org.apache.ambari.server.stack.upgrade.orchestrate.UpgradeContext;
import org.apache.ambari.server.stack.upgrade.orchestrate.UpgradeContextFactory;
import org.apache.ambari.server.stack.upgrade.orchestrate.UpgradeGroupHolder;
import org.apache.ambari.server.stack.upgrade.orchestrate.UpgradeHelper;
import org.apache.ambari.server.state.Cluster;
import org.apache.ambari.server.state.Clusters;
import org.apache.ambari.server.state.ConfigHelper;
import org.apache.ambari.server.state.Service;
import org.apache.ambari.server.state.ServiceComponent;
import org.apache.ambari.server.state.StackId;
import org.apache.ambari.server.state.StackInfo;
import org.apache.ambari.server.state.svccomphost.ServiceComponentHostServerActionEvent;
import org.apache.ambari.spi.upgrade.UpgradeAction;
import org.apache.ambari.spi.upgrade.UpgradeType;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.google.gson.JsonArray;
import com.google.gson.JsonObject;
import com.google.inject.Inject;
import com.google.inject.Provider;
import com.google.inject.assistedinject.Assisted;
import com.google.inject.persist.Transactional;

/**
 * Manages the ability to start and get status of upgrades.
 */
@StaticallyInject
public class UpgradeResourceProvider extends AbstractControllerResourceProvider {

  public static final String UPGRADE_CLUSTER_NAME = "Upgrade/cluster_name";
  public static final String UPGRADE_REPO_VERSION_ID = "Upgrade/repository_version_id";
  public static final String UPGRADE_TYPE = "Upgrade/upgrade_type";
  public static final String UPGRADE_PACK = "Upgrade/pack";
  public static final String UPGRADE_ID = "Upgrade/upgrade_id";
  public static final String UPGRADE_REQUEST_ID = "Upgrade/request_id";
  public static final String UPGRADE_ASSOCIATED_VERSION = "Upgrade/associated_version";
  public static final String UPGRADE_VERSIONS = "Upgrade/versions";
  public static final String UPGRADE_DIRECTION = "Upgrade/direction";
  public static final String UPGRADE_DOWNGRADE_ALLOWED = "Upgrade/downgrade_allowed";
  public static final String UPGRADE_REQUEST_STATUS = "Upgrade/request_status";
  public static final String UPGRADE_SUSPENDED = "Upgrade/suspended";
  public static final String UPGRADE_ABORT_REASON = "Upgrade/abort_reason";
  public static final String UPGRADE_SKIP_PREREQUISITE_CHECKS = "Upgrade/skip_prerequisite_checks";
  public static final String UPGRADE_FAIL_ON_CHECK_WARNINGS = "Upgrade/fail_on_check_warnings";

  /**
   * Skip slave/client component failures if the tasks are skippable.
   */
  public static final String UPGRADE_SKIP_FAILURES = "Upgrade/skip_failures";

  /**
   * Skip service check failures if the tasks are skippable.
   */
  public static final String UPGRADE_SKIP_SC_FAILURES = "Upgrade/skip_service_check_failures";

  /**
   * Skip manual verification tasks for hands-free upgrade/downgrade experience.
   */
  public static final String UPGRADE_SKIP_MANUAL_VERIFICATION = "Upgrade/skip_manual_verification";

  /**
   * When creating an upgrade of type {@link UpgradeType#HOST_ORDERED}, this
   * specifies the order in which the hosts are upgraded.
   * </p>
   *
   * <pre>
   * "host_order": [
   *   { "hosts":
   *       [ "c6401.ambari.apache.org, "c6402.ambari.apache.org", "c6403.ambari.apache.org" ],
   *     "service_checks": ["ZOOKEEPER"]
   *   },
   *   {
   *     "hosts": [ "c6404.ambari.apache.org, "c6405.ambari.apache.org"],
   *     "service_checks": ["ZOOKEEPER", "KAFKA"]
   *   }
   * ]
   * </pre>
   *
   */
  public static final String UPGRADE_HOST_ORDERED_HOSTS = "Upgrade/host_order";

  /**
   * Allows reversion of a successful upgrade of a patch.
   */
  public static final String UPGRADE_REVERT_UPGRADE_ID = "Upgrade/revert_upgrade_id";

  /**
   * The role that will be used when creating HRC's for the type
   * {@link StageWrapper.Type#UPGRADE_TASKS}.
   */
  protected static final String EXECUTE_TASK_ROLE = "ru_execute_tasks";

  /*
   * 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";

  protected 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<>(
      Arrays.asList(UPGRADE_REQUEST_ID, UPGRADE_CLUSTER_NAME));

  private static final Set<String> PROPERTY_IDS = new HashSet<>();

  private static final String DEFAULT_REASON_TEMPLATE = "Aborting upgrade %s";

  private static final Map<Resource.Type, String> KEY_PROPERTY_IDS = new HashMap<>();

  @Inject
  static UpgradeDAO s_upgradeDAO;

  @Inject
  private static Provider<AmbariMetaInfo> s_metaProvider = null;

  @Inject
  private static Provider<RequestFactory> s_requestFactory;

  @Inject
  private static Provider<StageFactory> s_stageFactory;

  @Inject
  private static Provider<Clusters> clusters = null;

  @Inject
  private static Provider<AmbariActionExecutionHelper> s_actionExecutionHelper;

  @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.
   */
  @Inject
  private static UpgradeHelper s_upgradeHelper;

  @Inject
  private static Configuration s_configuration;

  /**
   * Used to create instances of {@link UpgradeContext} with injected
   * dependencies.
   */
  @Inject
  private static UpgradeContextFactory s_upgradeContextFactory;

  @Inject
  private STOMPUpdatePublisher STOMPUpdatePublisher;

  @Inject
  private HostRoleCommandDAO hostRoleCommandDAO;

  @Inject
  private RequestDAO requestDAO;

  /**
   * Used for creating keytab regeneration stage inside of an upgrade request.
   */
  @Inject
  private static Provider<KerberosHelper> s_kerberosHelper;

  static {
    // properties
    PROPERTY_IDS.add(UPGRADE_CLUSTER_NAME);
    PROPERTY_IDS.add(UPGRADE_REPO_VERSION_ID);
    PROPERTY_IDS.add(UPGRADE_TYPE);
    PROPERTY_IDS.add(UPGRADE_PACK);
    PROPERTY_IDS.add(UPGRADE_ID);
    PROPERTY_IDS.add(UPGRADE_REQUEST_ID);
    PROPERTY_IDS.add(UPGRADE_ASSOCIATED_VERSION);
    PROPERTY_IDS.add(UPGRADE_VERSIONS);
    PROPERTY_IDS.add(UPGRADE_DIRECTION);
    PROPERTY_IDS.add(UPGRADE_DOWNGRADE_ALLOWED);
    PROPERTY_IDS.add(UPGRADE_SUSPENDED);
    PROPERTY_IDS.add(UPGRADE_SKIP_FAILURES);
    PROPERTY_IDS.add(UPGRADE_SKIP_SC_FAILURES);
    PROPERTY_IDS.add(UPGRADE_SKIP_MANUAL_VERIFICATION);
    PROPERTY_IDS.add(UPGRADE_SKIP_PREREQUISITE_CHECKS);
    PROPERTY_IDS.add(UPGRADE_FAIL_ON_CHECK_WARNINGS);
    PROPERTY_IDS.add(UPGRADE_HOST_ORDERED_HOSTS);
    PROPERTY_IDS.add(UPGRADE_REVERT_UPGRADE_ID);

    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);
    KEY_PROPERTY_IDS.put(Resource.Type.Cluster, UPGRADE_CLUSTER_NAME);
  }

  private static final Logger LOG = LoggerFactory.getLogger(UpgradeResourceProvider.class);

  /**
   * Constructor.
   *
   * @param controller
   *          the controller
   */
  @Inject
  public UpgradeResourceProvider(@Assisted AmbariManagementController controller) {
    super(Resource.Type.Upgrade, PROPERTY_IDS, KEY_PROPERTY_IDS, controller);
  }

  @Override
  public RequestStatus createResources(final Request request) throws SystemException,
      UnsupportedPropertyException, ResourceAlreadyExistsException, NoSuchParentResourceException {

    Set<Map<String, Object>> requestMaps = request.getProperties();

    if (requestMaps.size() > 1) {
      throw new SystemException("Can only initiate one upgrade per request.");
    }

    // !!! above check ensures only one
    final Map<String, Object> requestMap = requestMaps.iterator().next();
    final String clusterName = (String) requestMap.get(UPGRADE_CLUSTER_NAME);
    final Cluster cluster;

    try {
      cluster = clusters.get().getCluster(clusterName);
    } catch (AmbariException e) {
      throw new NoSuchParentResourceException(
          String.format("Cluster %s could not be loaded", clusterName));
    }

    if (!AuthorizationHelper.isAuthorized(ResourceType.CLUSTER, cluster.getResourceId(),
        EnumSet.of(RoleAuthorization.CLUSTER_UPGRADE_DOWNGRADE_STACK))) {
      throw new AuthorizationException("The authenticated user does not have authorization to " +
          "manage upgrade and downgrade");
    }

    UpgradeEntity entity = createResources(new Command<UpgradeEntity>() {
      @Override
      public UpgradeEntity invoke() throws AmbariException, AuthorizationException {

        // create the context, validating the properties in the process
        final UpgradeContext upgradeContext = s_upgradeContextFactory.create(cluster, requestMap);

        try {
          return createUpgrade(upgradeContext);
        } catch (Exception e) {
          LOG.error("Error appears during upgrade task submitting", e);

          // Any error caused in the createUpgrade will initiate transaction
          // rollback
          // As we operate inside with cluster data, any cache which belongs to
          // cluster need to be flushed
          clusters.get().invalidate(cluster);
          throw e;
        }
      }
    });

    if (null == entity) {
      throw new SystemException("Could not load upgrade");
    }

    notifyCreate(Resource.Type.Upgrade, request);

    Resource res = new ResourceImpl(Resource.Type.Upgrade);
    res.setProperty(UPGRADE_REQUEST_ID, entity.getRequestId());
    return new RequestStatusImpl(null, Collections.singleton(res));
  }

  @Override
  public Set<Resource> getResources(Request request, Predicate predicate) throws SystemException,
      UnsupportedPropertyException, NoSuchResourceException, NoSuchParentResourceException {

    Set<Resource> results = new HashSet<>();
    Set<String> requestPropertyIds = getRequestPropertyIds(request, predicate);

    for (Map<String, Object> propertyMap : getPropertyMaps(predicate)) {
      String clusterName = (String) propertyMap.get(UPGRADE_CLUSTER_NAME);

      if (null == clusterName || clusterName.isEmpty()) {
        throw new IllegalArgumentException(
            "The cluster name is required when querying for upgrades");
      }

      Cluster cluster;
      try {
        cluster = clusters.get().getCluster(clusterName);
      } catch (AmbariException e) {
        throw new NoSuchResourceException(
            String.format("Cluster %s could not be loaded", clusterName));
      }

      List<UpgradeEntity> upgrades = new ArrayList<>();

      String upgradeIdStr = (String) propertyMap.get(UPGRADE_REQUEST_ID);
      if (null != upgradeIdStr) {
        UpgradeEntity upgrade = s_upgradeDAO.findUpgradeByRequestId(Long.valueOf(upgradeIdStr));

        if (null != upgrade) {
          upgrades.add(upgrade);
        }
      } else {
        upgrades = s_upgradeDAO.findUpgrades(cluster.getClusterId());
      }

      for (UpgradeEntity entity : upgrades) {
        Resource r = toResource(entity, clusterName, requestPropertyIds);
        results.add(r);

        RequestEntity rentity = s_requestDAO.findByPK(entity.getRequestId());

        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());

        if (calc.getStatus() == HostRoleStatus.ABORTED && entity.isSuspended()) {
          double percent = calculateAbortedProgress(summary);
          setResourceProperty(r, REQUEST_PROGRESS_PERCENT_ID, percent*100, requestPropertyIds);
        } else {
          setResourceProperty(r, REQUEST_PROGRESS_PERCENT_ID, calc.getPercent(), requestPropertyIds);
        }

        setResourceProperty(r, REQUEST_STATUS_PROPERTY_ID, calc.getStatus(), requestPropertyIds);
      }
    }

    return results;
  }

  /**
   * Unlike in CalculatedStatus, we can't use ABORTED here as a COMPLETED state.
   * Therefore, the values will be slightly off since in CalulatedStatus, ABORTED
   * contributes all of its progress to the overall progress, but here it
   * contributes none of it.
   *
   * Since this is specifically for ABORTED upgrades that are
   * also suspended, the percentages should come out pretty close after ABORTED move back
   * to PENDING.
   *
   * @return the percent complete, counting ABORTED as zero percent.
   */
  public static double calculateAbortedProgress(Map<Long, HostRoleCommandStatusSummaryDTO> summary) {
    // !!! use the raw states to determine percent completes
    Map<HostRoleStatus, Integer> countTotals = new HashMap<>();
    int totalTasks = 0;


    for (HostRoleCommandStatusSummaryDTO statusSummary : summary.values()) {
      totalTasks += statusSummary.getTaskTotal();
      for (Map.Entry<HostRoleStatus, Integer> entry : statusSummary.getCounts().entrySet()) {
        if (!countTotals.containsKey(entry.getKey())) {
          countTotals.put(entry.getKey(), Integer.valueOf(0));
        }
        countTotals.put(entry.getKey(), countTotals.get(entry.getKey()) + entry.getValue());
      }
    }

    double percent = 0d;

    for (HostRoleStatus status : HostRoleStatus.values()) {
      if (!countTotals.containsKey(status)) {
        countTotals.put(status, Integer.valueOf(0));
      }
      double countValue = countTotals.get(status);

      // !!! calculation lifted from CalculatedStatus
      switch (status) {
        case ABORTED:
          // !!! see javadoc
          break;
        case HOLDING:
        case HOLDING_FAILED:
        case HOLDING_TIMEDOUT:
        case IN_PROGRESS:
        case PENDING:  // shouldn't be any, we're supposed to be ABORTED
          percent += countValue * 0.35d;
          break;
        case QUEUED:
          percent += countValue * 0.09d;
          break;
        default:
          if (status.isCompletedState()) {
            percent += countValue / totalTasks;
          }
          break;
      }
    }

    return percent;
  }

  @Override
  public RequestStatus updateResources(final Request request, Predicate predicate)
      throws SystemException, UnsupportedPropertyException, NoSuchResourceException,
      NoSuchParentResourceException {

    Set<Map<String, Object>> requestMaps = request.getProperties();

    if (requestMaps.size() > 1) {
      throw new SystemException("Can only update one upgrade per request.");
    }

    // !!! above check ensures only one
    final Map<String, Object> propertyMap = requestMaps.iterator().next();

    final String clusterName = (String) propertyMap.get(UPGRADE_CLUSTER_NAME);
    final Cluster cluster;

    try {
      cluster = clusters.get().getCluster(clusterName);
    } catch (AmbariException e) {
      throw new NoSuchParentResourceException(
          String.format("Cluster %s could not be loaded", clusterName));
    }

    if (!AuthorizationHelper.isAuthorized(ResourceType.CLUSTER, cluster.getResourceId(),
        EnumSet.of(RoleAuthorization.CLUSTER_UPGRADE_DOWNGRADE_STACK))) {
      throw new AuthorizationException("The authenticated user does not have authorization to " +
          "manage upgrade and downgrade");
    }

    String requestIdProperty = (String) propertyMap.get(UPGRADE_REQUEST_ID);
    if (null == requestIdProperty) {
      throw new IllegalArgumentException(String.format("%s is required", UPGRADE_REQUEST_ID));
    }

    long requestId = Long.parseLong(requestIdProperty);
    UpgradeEntity upgradeEntity = s_upgradeDAO.findUpgradeByRequestId(requestId);
    if( null == upgradeEntity){
      String exceptionMessage = MessageFormat.format("The upgrade with request ID {0} was not found", requestIdProperty);
      throw new NoSuchParentResourceException(exceptionMessage);
    }

    // the properties which are allowed to be updated; the request must include
    // at least 1
    List<String> updatableProperties = Lists.newArrayList(UPGRADE_REQUEST_STATUS,
        UPGRADE_SKIP_FAILURES, UPGRADE_SKIP_SC_FAILURES);

    boolean isRequiredPropertyInRequest = CollectionUtils.containsAny(updatableProperties,
        propertyMap.keySet());

    if (!isRequiredPropertyInRequest) {
      String exceptionMessage = MessageFormat.format(
          "At least one of the following properties is required in the request: {0}",
          StringUtils.join(updatableProperties, ", "));
      throw new IllegalArgumentException(exceptionMessage);
    }

    String requestStatus = (String) propertyMap.get(UPGRADE_REQUEST_STATUS);
    String skipFailuresRequestProperty = (String) propertyMap.get(UPGRADE_SKIP_FAILURES);
    String skipServiceCheckFailuresRequestProperty = (String) propertyMap.get(UPGRADE_SKIP_SC_FAILURES);

    if (null != requestStatus) {
      HostRoleStatus status = HostRoleStatus.valueOf(requestStatus);

      // When aborting an upgrade, the suspend flag must be present to indicate
      // if the upgrade is merely being paused (suspended=true) or aborted to initiate a downgrade (suspended=false).
      boolean suspended = false;
      if (status == HostRoleStatus.ABORTED && !propertyMap.containsKey(UPGRADE_SUSPENDED)){
        throw new IllegalArgumentException(String.format(
            "When changing the state of an upgrade to %s, the %s property is required to be either true or false.",
            status, UPGRADE_SUSPENDED ));
      } else if (status == HostRoleStatus.ABORTED) {
        suspended = Boolean.valueOf((String) propertyMap.get(UPGRADE_SUSPENDED));
      }

      try {
        setUpgradeRequestStatus(cluster, requestId, status, suspended, propertyMap);
      } catch (AmbariException ambariException) {
        throw new SystemException(ambariException.getMessage(), ambariException);
      }
    }

    // if either of the skip failure settings are in the request, then we need
    // to iterate over the entire series of tasks anyway, so do them both at the
    // same time
    if (StringUtils.isNotEmpty(skipFailuresRequestProperty)
        || StringUtils.isNotEmpty(skipServiceCheckFailuresRequestProperty)) {
      // grab the current settings for both
      boolean skipFailures = upgradeEntity.isComponentFailureAutoSkipped();
      boolean skipServiceCheckFailures = upgradeEntity.isServiceCheckFailureAutoSkipped();

      if (null != skipFailuresRequestProperty) {
        skipFailures = Boolean.parseBoolean(skipFailuresRequestProperty);
      }

      if (null != skipServiceCheckFailuresRequestProperty) {
        skipServiceCheckFailures = Boolean.parseBoolean(skipServiceCheckFailuresRequestProperty);
      }

      s_hostRoleCommandDAO.updateAutomaticSkipOnFailure(requestId, skipFailures, skipServiceCheckFailures);

      upgradeEntity.setAutoSkipComponentFailures(skipFailures);
      upgradeEntity.setAutoSkipServiceCheckFailures(skipServiceCheckFailures);
      s_upgradeDAO.merge(upgradeEntity);

    }

    return getRequestStatus(null);
  }

  @Override
  public RequestStatus deleteResources(Request request, Predicate predicate) throws SystemException,
      UnsupportedPropertyException, NoSuchResourceException, NoSuchParentResourceException {
    throw new SystemException("Cannot delete Upgrades");
  }

  @Override
  protected Set<String> getPKPropertyIds() {
    return PK_PROPERTY_IDS;
  }

  private Resource toResource(UpgradeEntity entity, String clusterName, Set<String> requestedIds) {
    ResourceImpl resource = new ResourceImpl(Resource.Type.Upgrade);

    setResourceProperty(resource, UPGRADE_ID, entity.getId(), requestedIds);
    setResourceProperty(resource, UPGRADE_CLUSTER_NAME, clusterName, requestedIds);
    setResourceProperty(resource, UPGRADE_TYPE, entity.getUpgradeType(), requestedIds);
    setResourceProperty(resource, UPGRADE_PACK, entity.getUpgradePackage(), requestedIds);
    setResourceProperty(resource, UPGRADE_REQUEST_ID, entity.getRequestId(), requestedIds);
    setResourceProperty(resource, UPGRADE_DIRECTION, entity.getDirection(), requestedIds);
    setResourceProperty(resource, UPGRADE_SUSPENDED, entity.isSuspended(), requestedIds);
    setResourceProperty(resource, UPGRADE_DOWNGRADE_ALLOWED, entity.isDowngradeAllowed(), requestedIds);
    setResourceProperty(resource, UPGRADE_SKIP_FAILURES, entity.isComponentFailureAutoSkipped(), requestedIds);
    setResourceProperty(resource, UPGRADE_SKIP_SC_FAILURES, entity.isServiceCheckFailureAutoSkipped(), requestedIds);

    // set the assocaited to/from version (to/from is dictated by direction)
    RepositoryVersionEntity repositoryVersion = entity.getRepositoryVersion();
    setResourceProperty(resource, UPGRADE_ASSOCIATED_VERSION, repositoryVersion.getVersion(), requestedIds);

    // now set the target verison for all services in the upgrade
    Map<String, RepositoryVersions> repositoryVersions = new HashMap<>();
    for( UpgradeHistoryEntity history : entity.getHistory() ){
      RepositoryVersions serviceVersions = repositoryVersions.get(history.getServiceName());
      if (null != serviceVersions) {
        continue;
      }

      serviceVersions = new RepositoryVersions(history.getFromReposistoryVersion(),
          history.getTargetRepositoryVersion());

      repositoryVersions.put(history.getServiceName(), serviceVersions);
    }

    setResourceProperty(resource, UPGRADE_VERSIONS, repositoryVersions, requestedIds);

    return resource;
  }

  /**
   * Inject variables into the
   * {@link org.apache.ambari.server.orm.entities.UpgradeItemEntity}, whose
   * tasks may use strings like {{configType/propertyName}} that need to be
   * retrieved from the properties.
   *
   * @param configHelper
   *          Configuration Helper
   * @param cluster
   *          Cluster
   * @param upgradeItem
   *          the item whose tasks will be injected.
   */
  private void injectVariables(ConfigHelper configHelper, Cluster cluster,
      UpgradeItemEntity upgradeItem) {
    final String regexp = "(\\{\\{.*?\\}\\})";

    String task = upgradeItem.getTasks();
    if (task != null && !task.isEmpty()) {
      Matcher m = Pattern.compile(regexp).matcher(task);
      while (m.find()) {
        String origVar = m.group(1);
        String configValue = configHelper.getPlaceholderValueFromDesiredConfigurations(cluster,
            origVar);

        if (null != configValue) {
          task = task.replace(origVar, configValue);
        } else {
          LOG.error("Unable to retrieve value for {}", origVar);
        }

      }
      upgradeItem.setTasks(task);
    }
  }

  /**
   * Creates the upgrade. All Request/Stage/Task and Upgrade entities will exist
   * in the database when this method completes.
   * <p/>
   * This method itself must not be wrapped in a transaction since it can
   * potentially make 1000's of database calls while building the entities
   * before persisting them. This would create a long-lived transaction which
   * could lead to database deadlock issues. Instead, only the creation of the
   * actual entities is wrapped in a {@link Transactional} block.
   *
   * @param upgradeContext
   * @return
   * @throws AmbariException
   * @throws AuthorizationException
   */
  protected UpgradeEntity createUpgrade(UpgradeContext upgradeContext)
      throws AmbariException, AuthorizationException {

    UpgradePack pack = upgradeContext.getUpgradePack();
    Cluster cluster = upgradeContext.getCluster();
    Direction direction = upgradeContext.getDirection();

    ConfigHelper configHelper = getManagementController().getConfigHelper();

    List<UpgradeGroupHolder> groups = s_upgradeHelper.createSequence(pack, upgradeContext);

    if (groups.isEmpty()) {
      throw new AmbariException("There are no groupings available");
    }

    // Non Rolling Upgrades require a group with name "UPDATE_DESIRED_REPOSITORY_ID".
    // This is needed as a marker to indicate which version to use when an upgrade is paused.
    if (pack.getType() == UpgradeType.NON_ROLLING) {
      boolean foundUpdateDesiredRepositoryIdGrouping = false;
      for (UpgradeGroupHolder group : groups) {
        if (group.groupClass == UpdateStackGrouping.class) {
          foundUpdateDesiredRepositoryIdGrouping = true;
          break;
        }
      }

      if (!foundUpdateDesiredRepositoryIdGrouping) {
        throw new AmbariException(String.format(
            "Express upgrade packs are required to have a group of type %s. The upgrade pack %s is missing this grouping.",
            "update-stack", pack.getName()));
      }
    }

    List<UpgradeGroupEntity> groupEntities = new ArrayList<>();
    RequestStageContainer req = createRequest(upgradeContext);

    UpgradeEntity upgrade = new UpgradeEntity();
    upgrade.setRepositoryVersion(upgradeContext.getRepositoryVersion());
    upgrade.setClusterId(cluster.getClusterId());
    upgrade.setDirection(direction);
    upgrade.setUpgradePackage(pack.getName());
    upgrade.setUpgradePackStackId(pack.getOwnerStackId());
    upgrade.setUpgradeType(pack.getType());
    upgrade.setAutoSkipComponentFailures(upgradeContext.isComponentFailureAutoSkipped());
    upgrade.setAutoSkipServiceCheckFailures(upgradeContext.isServiceCheckFailureAutoSkipped());
    upgrade.setDowngradeAllowed(upgradeContext.isDowngradeAllowed());
    upgrade.setOrchestration(upgradeContext.getOrchestrationType());

    // create to/from history for this upgrade - this should be done before any
    // possible changes to the desired version for components
    addComponentHistoryToUpgrade(cluster, upgrade, upgradeContext);

    /*
    During a Rolling Upgrade, change the desired Stack Id if jumping across
