/*
 * 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
 * <p/>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p/>
 * 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.topology;

import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;

import javax.annotation.Nullable;
import javax.inject.Inject;

import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.ClusterNotFoundException;
import org.apache.ambari.server.DuplicateResourceException;
import org.apache.ambari.server.Role;
import org.apache.ambari.server.RoleCommand;
import org.apache.ambari.server.actionmanager.HostRoleCommand;
import org.apache.ambari.server.actionmanager.HostRoleCommandFactory;
import org.apache.ambari.server.actionmanager.HostRoleStatus;
import org.apache.ambari.server.agent.stomp.HostLevelParamsHolder;
import org.apache.ambari.server.controller.AmbariManagementController;
import org.apache.ambari.server.controller.AmbariServer;
import org.apache.ambari.server.controller.ClusterRequest;
import org.apache.ambari.server.controller.ConfigGroupRequest;
import org.apache.ambari.server.controller.ConfigurationRequest;
import org.apache.ambari.server.controller.RequestStatusResponse;
import org.apache.ambari.server.controller.RootComponent;
import org.apache.ambari.server.controller.ServiceComponentHostRequest;
import org.apache.ambari.server.controller.ServiceComponentRequest;
import org.apache.ambari.server.controller.ServiceRequest;
import org.apache.ambari.server.controller.internal.AbstractResourceProvider;
import org.apache.ambari.server.controller.internal.ComponentResourceProvider;
import org.apache.ambari.server.controller.internal.ConfigGroupResourceProvider;
import org.apache.ambari.server.controller.internal.HostComponentResourceProvider;
import org.apache.ambari.server.controller.internal.HostResourceProvider;
import org.apache.ambari.server.controller.internal.ProvisionClusterRequest;
import org.apache.ambari.server.controller.internal.RequestImpl;
import org.apache.ambari.server.controller.internal.ServiceResourceProvider;
import org.apache.ambari.server.controller.internal.Stack;
import org.apache.ambari.server.controller.internal.VersionDefinitionResourceProvider;
import org.apache.ambari.server.controller.predicate.EqualsPredicate;
import org.apache.ambari.server.controller.spi.ClusterController;
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.utilities.ClusterControllerHelper;
import org.apache.ambari.server.orm.dao.RepositoryVersionDAO;
import org.apache.ambari.server.orm.entities.RepositoryVersionEntity;
import org.apache.ambari.server.security.authorization.AuthorizationException;
import org.apache.ambari.server.state.Cluster;
import org.apache.ambari.server.state.Clusters;
import org.apache.ambari.server.state.Config;
import org.apache.ambari.server.state.ConfigFactory;
import org.apache.ambari.server.state.ConfigHelper;
import org.apache.ambari.server.state.DesiredConfig;
import org.apache.ambari.server.state.Host;
import org.apache.ambari.server.state.SecurityType;
import org.apache.ambari.server.state.StackId;
import org.apache.ambari.server.state.configgroup.ConfigGroup;
import org.apache.ambari.server.utils.RetryHelper;
import org.apache.ambari.spi.RepositoryType;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Function;
import com.google.common.collect.Collections2;
import com.google.common.collect.Iterables;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.Striped;
import com.google.inject.Provider;


/**
 * Provides topology related information as well as access to the core Ambari functionality.
 */
public class AmbariContext {

  public enum TaskType {INSTALL, START}

  @Inject
  private PersistedState persistedState;

  /**
   * Used for creating read-only instances of existing {@link Config} in order
   * to send them to the {@link ConfigGroupResourceProvider} to create
   * {@link ConfigGroup}s.
   */
  @Inject
  ConfigFactory configFactory;

  @Inject
  RepositoryVersionDAO repositoryVersionDAO;

  /**
   * Used for getting configuration property values from stack and services.
   */
  @Inject
  private Provider<ConfigHelper> configHelper;

  @Inject
  HostLevelParamsHolder hostLevelParamsHolder;

  private static AmbariManagementController controller;
  private static ClusterController clusterController;
  //todo: task id's.  Use existing mechanism for getting next task id sequence
  private final static AtomicLong nextTaskId = new AtomicLong(10000);

  private static HostRoleCommandFactory hostRoleCommandFactory;
  private static HostResourceProvider hostResourceProvider;
  private static ServiceResourceProvider serviceResourceProvider;
  private static ComponentResourceProvider componentResourceProvider;
  private static HostComponentResourceProvider hostComponentResourceProvider;
  private static VersionDefinitionResourceProvider versionDefinitionResourceProvider;

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


  /**
   * When config groups are created using Blueprints these are created when
   * hosts join a hostgroup and are added to the corresponding config group.
   * Since hosts join in parallel there might be a race condition in creating
   * the config group a host is to be added to. Thus we need to synchronize
   * the creation of config groups with the same name.
   */
  private Striped<Lock> configGroupCreateLock = Striped.lazyWeakLock(1);

  public boolean isClusterKerberosEnabled(long clusterId) {
    Cluster cluster;
    try {
      cluster = getController().getClusters().getClusterById(clusterId);
    } catch (AmbariException e) {
      throw new RuntimeException("Parent Cluster resource doesn't exist.  clusterId= " + clusterId);
    }
    return cluster.getSecurityType() == SecurityType.KERBEROS;
  }

  //todo: change return type to a topology abstraction
  public HostRoleCommand createAmbariTask(long requestId, long stageId, String component, String host,
                                          TaskType type, boolean skipFailure) {
    HostRoleCommand task = hostRoleCommandFactory.create(
            host, Role.valueOf(component), null, RoleCommand.valueOf(type.name()), false, skipFailure);
    task.setStatus(HostRoleStatus.PENDING);
    task.setCommandDetail(String.format("Logical Task: %s component %s on host %s", type.name(), component, host));
    task.setTaskId(nextTaskId.getAndIncrement());
    task.setRequestId(requestId);
    task.setStageId(stageId);

    return task;
  }

  //todo: change return type to a topology abstraction
  public HostRoleCommand createAmbariTask(long taskId, long requestId, long stageId,
                                          String component, String host, TaskType type, boolean skipFailure) {
    synchronized (nextTaskId) {
      if (nextTaskId.get() <= taskId) {
        nextTaskId.set(taskId + 1);
      }
    }

    HostRoleCommand task = hostRoleCommandFactory.create(
        host, Role.valueOf(component), null, RoleCommand.valueOf(type.name()), false, skipFailure);
    task.setStatus(HostRoleStatus.PENDING);
    task.setCommandDetail(String.format("Logical Task: %s component %s on host %s",
        type.name(), component, host));
    task.setTaskId(taskId);
    task.setRequestId(requestId);
    task.setStageId(stageId);

    return task;
  }

  public HostRoleCommand getPhysicalTask(long id) {
    return getController().getActionManager().getTaskById(id);
  }

  public Collection<HostRoleCommand> getPhysicalTasks(Collection<Long> ids) {
    return getController().getActionManager().getTasks(ids);
  }

  public void createAmbariResources(ClusterTopology topology, String clusterName, SecurityType securityType,
                                    String repoVersionString, Long repoVersionId) {
    Stack stack = topology.getBlueprint().getStack();
    StackId stackId = new StackId(stack.getName(), stack.getVersion());

    RepositoryVersionEntity repoVersion = null;
    if (StringUtils.isEmpty(repoVersionString) && null == repoVersionId) {
      List<RepositoryVersionEntity> stackRepoVersions = repositoryVersionDAO.findByStack(stackId);

      if (stackRepoVersions.isEmpty()) {
        // !!! no repos, try to get the version for the stack
        VersionDefinitionResourceProvider vdfProvider = getVersionDefinitionResourceProvider();

        Map<String, Object> properties = new HashMap<>();
        properties.put(VersionDefinitionResourceProvider.VERSION_DEF_AVAILABLE_DEFINITION, stackId.toString());

        Request request = new RequestImpl(Collections.<String>emptySet(),
            Collections.singleton(properties), Collections.<String, String>emptyMap(), null);

        Long defaultRepoVersionId = null;

        try {
          RequestStatus requestStatus = vdfProvider.createResources(request);
          if (!requestStatus.getAssociatedResources().isEmpty()) {
            Resource resource = requestStatus.getAssociatedResources().iterator().next();
            defaultRepoVersionId = (Long) resource.getPropertyValue(VersionDefinitionResourceProvider.VERSION_DEF_ID);
          }
        } catch (Exception e) {
          throw new IllegalArgumentException(String.format(
              "Failed to create a default repository version definition for stack %s. "
              + "This typically is a result of not loading the stack correctly or being able "
              + "to load information about released versions.  Create a repository version "
              + " and try again.", stackId), e);
        }

        repoVersion = repositoryVersionDAO.findByPK(defaultRepoVersionId);
        // !!! better not!
        if (null == repoVersion) {
          throw new IllegalArgumentException(String.format(
              "Failed to load the default repository version definition for stack %s. "
              + "Check for a valid repository version and try again.", stackId));
        }

      } else if (stackRepoVersions.size() > 1) {

        Function<RepositoryVersionEntity, String> function = new Function<RepositoryVersionEntity, String>() {
          @Override
          public String apply(RepositoryVersionEntity input) {
            return input.getVersion();
          }
        };

        Collection<String> versions = Collections2.transform(stackRepoVersions, function);

        throw new IllegalArgumentException(String.format("Several repositories were found for %s:  %s.  Specify the version"
            + " with '%s'", stackId, StringUtils.join(versions, ", "), ProvisionClusterRequest.REPO_VERSION_PROPERTY));
      } else {
        repoVersion = stackRepoVersions.get(0);
        LOG.warn("Cluster is being provisioned using the single matching repository version {}", repoVersion.getVersion());
      }
    } else {
      if (null != repoVersionId) {
        repoVersion = repositoryVersionDAO.findByPK(repoVersionId);

        if (null == repoVersion) {
          throw new IllegalArgumentException(String.format(
            "Could not identify repository version with repository version id %s for installing services. "
              + "Specify a valid repository version id with '%s'",
            repoVersionId, ProvisionClusterRequest.REPO_VERSION_ID_PROPERTY));
        }
      } else {
        repoVersion = repositoryVersionDAO.findByStackAndVersion(stackId, repoVersionString);

        if (null == repoVersion) {
          throw new IllegalArgumentException(String.format(
            "Could not identify repository version with stack %s and version %s for installing services. "
              + "Specify a valid version with '%s'",
            stackId, repoVersionString, ProvisionClusterRequest.REPO_VERSION_PROPERTY));
        }
      }

      if (!Objects.equals(repoVersion.getStackId(), stackId)) {
        String repoVersionPair = repoVersionId != null
          ? String.format("'%s' = %d", ProvisionClusterRequest.REPO_VERSION_ID_PROPERTY, repoVersionId)
          : String.format("'%s' = '%s'", ProvisionClusterRequest.REPO_VERSION_PROPERTY, repoVersionString);
        String msg = String.format(
          "The stack specified in the blueprint (%s) and the repository version (%s for %s) should match",
          stackId, repoVersion.getStackId(), repoVersionPair);
        throw new IllegalArgumentException(msg);
      }
    }

    // only use a STANDARD repo when creating a new cluster
    if (repoVersion.getType() != RepositoryType.STANDARD) {
      throw new IllegalArgumentException(String.format(
          "Unable to create a cluster using the following repository since it is not a STANDARD type: %s",
          repoVersion));
    }

    createAmbariClusterResource(clusterName, stack.getName(), stack.getVersion(), securityType);
    createAmbariServiceAndComponentResources(topology, clusterName, stackId, repoVersion.getId());
  }

  public void createAmbariClusterResource(String clusterName, String stackName, String stackVersion, SecurityType securityType) {
    String stackInfo = String.format("%s-%s", stackName, stackVersion);
    final ClusterRequest clusterRequest = new ClusterRequest(null, clusterName, null, securityType, stackInfo, null);

    try {
      RetryHelper.executeWithRetry(new Callable<Object>() {
        @Override
        public Object call() throws Exception {
          getController().createCluster(clusterRequest);
          return null;
        }
      });

    } catch (AmbariException e) {
      LOG.error("Failed to create Cluster resource: ", e);
      if (e.getCause() instanceof DuplicateResourceException) {
        throw new IllegalArgumentException(e);
      } else {
        throw new RuntimeException("Failed to create Cluster resource: " + e, e);
      }
    }
  }

  public void createAmbariServiceAndComponentResources(ClusterTopology topology, String clusterName,
      StackId stackId, Long repositoryVersionId) {
    Collection<String> services = topology.getBlueprint().getServices();

    try {
      Cluster cluster = getController().getClusters().getCluster(clusterName);
      services.removeAll(cluster.getServices().keySet());
    } catch (AmbariException e) {
      throw new RuntimeException("Failed to persist service and component resources: " + e, e);
    }
    Set<ServiceRequest> serviceRequests = new HashSet<>();
    Set<ServiceComponentRequest> componentRequests = new HashSet<>();
    for (String service : services) {
      String credentialStoreEnabled = topology.getBlueprint().getCredentialStoreEnabled(service);
      serviceRequests.add(new ServiceRequest(clusterName, service, repositoryVersionId, null, credentialStoreEnabled));

      for (String component : topology.getBlueprint().getComponents(service)) {
        String recoveryEnabled = topology.getBlueprint().getRecoveryEnabled(service, component);
        componentRequests.add(new ServiceComponentRequest(clusterName, service, component, null, recoveryEnabled));
      }
    }
    try {
      getServiceResourceProvider().createServices(serviceRequests);
      getComponentResourceProvider().createComponents(componentRequests);
    } catch (AmbariException | AuthorizationException e) {
      throw new RuntimeException("Failed to persist service and component resources: " + e, e);
    }
    // set all services state to INSTALLED->STARTED
    // this is required so the user can start failed services at the service level
    Map<String, Object> installProps = new HashMap<>();
    installProps.put(ServiceResourceProvider.SERVICE_SERVICE_STATE_PROPERTY_ID, "INSTALLED");
    installProps.put(ServiceResourceProvider.SERVICE_CLUSTER_NAME_PROPERTY_ID, clusterName);
    Map<String, Object> startProps = new HashMap<>();
    startProps.put(ServiceResourceProvider.SERVICE_SERVICE_STATE_PROPERTY_ID, "STARTED");
    startProps.put(ServiceResourceProvider.SERVICE_CLUSTER_NAME_PROPERTY_ID, clusterName);
    Predicate predicate = new EqualsPredicate<>(
      ServiceResourceProvider.SERVICE_CLUSTER_NAME_PROPERTY_ID, clusterName);
    try {
      getServiceResourceProvider().updateResources(
          new RequestImpl(null, Collections.singleton(installProps), null, null), predicate);

      getServiceResourceProvider().updateResources(
        new RequestImpl(null, Collections.singleton(startProps), null, null), predicate);
    } catch (Exception e) {
      // just log as this won't prevent cluster from being provisioned correctly
      LOG.error("Unable to update state of services during cluster provision: " + e, e);
    }
  }

  public void createAmbariHostResources(long  clusterId, String hostName, Map<String, Collection<String>> components)  {
    Host host;
    try {
      host = getController().getClusters().getHost(hostName);
    } catch (AmbariException e) {
      // system exception, shouldn't occur
      throw new RuntimeException(String.format(
          "Unable to obtain host instance '%s' when persisting host resources", hostName));
    }

    Cluster cluster = null;
    try {
      cluster = getController().getClusters().getClusterById(clusterId);
    } catch (AmbariException e) {
      LOG.error("Cannot get cluster for clusterId = " + clusterId, e);
      throw new RuntimeException(e);
    }
    String clusterName = cluster.getClusterName();

    Map<String, Object> properties = new HashMap<>();
    properties.put(HostResourceProvider.HOST_CLUSTER_NAME_PROPERTY_ID, clusterName);
    properties.put(HostResourceProvider.HOST_HOST_NAME_PROPERTY_ID, hostName);
    properties.put(HostResourceProvider.HOST_RACK_INFO_PROPERTY_ID, host.getRackInfo());

    try {
      getHostResourceProvider().createHosts(new RequestImpl(null, Collections.singleton(properties), null, null));
    } catch (AmbariException | AuthorizationException e) {
      LOG.error("Unable to create host component resource for host {}", hostName, e);
      throw new RuntimeException(String.format("Unable to create host resource for host '%s': %s",
          hostName, e.toString()), e);
    }

    final Set<ServiceComponentHostRequest> requests = new HashSet<>();

    for (Map.Entry<String, Collection<String>> entry : components.entrySet()) {
      String service = entry.getKey();
      for (String component : entry.getValue()) {
        //todo: handle this in a generic manner.  These checks are all over the code
        try {
          if (cluster.getService(service) != null && !component.equals(RootComponent.AMBARI_SERVER.name())) {
            requests.add(new ServiceComponentHostRequest(clusterName, service, component, hostName, null));
          }
        } catch(AmbariException se) {
          LOG.warn("Service already deleted from cluster: {}", service);
        }
      }
    }
    try {
      RetryHelper.executeWithRetry(new Callable<Object>() {
        @Override
        public Object call() throws Exception {
          getController().createHostComponents(requests, true);
          return null;
        }
      });
      hostLevelParamsHolder.updateData(hostLevelParamsHolder.getCurrentData(host.getHostId()));
    } catch (AmbariException e) {
      LOG.error("Unable to create host component resource for host {}", hostName, e);
      throw new RuntimeException(String.format("Unable to create host component resource for host '%s': %s",
          hostName, e.toString()), e);
    }
  }

  public Long getNextRequestId() {
    return getController().getActionManager().getNextRequestId();
  }

  public synchronized static AmbariManagementController getController() {
    if (controller == null) {
      controller = AmbariServer.getController();
    }
    return controller;
  }

  public synchronized static ClusterController getClusterController() {
    if (clusterController == null) {
      clusterController = ClusterControllerHelper.getClusterController();
    }
    return clusterController;
  }

  public static void init(HostRoleCommandFactory factory) {
    hostRoleCommandFactory = factory;
  }

  public void registerHostWithConfigGroup(final String hostName, final ClusterTopology topology, final String groupName) {
    String qualifiedGroupName = getConfigurationGroupName(topology.getBlueprint().getName(), groupName);

    Lock configGroupLock = configGroupCreateLock.get(qualifiedGroupName);

    try {
      configGroupLock.lock();

      boolean hostAdded = RetryHelper.executeWithRetry(new Callable<Boolean>() {
        @Override
        public Boolean call() throws Exception {
          return addHostToExistingConfigGroups(hostName, topology, qualifiedGroupName);
        }
      });
      if (!hostAdded) {
        createConfigGroupsAndRegisterHost(topology, groupName);
      }
    } catch (Exception e) {
      LOG.error("Unable to register config group for host: ", e);
      throw new RuntimeException("Unable to register config group for host: " + hostName);
    }
    finally {
      configGroupLock.unlock();
    }
  }

  public RequestStatusResponse installHost(String hostName, String clusterName, Collection<String> skipInstallForComponents, Collection<String> dontSkipInstallForComponents, boolean skipFailure) {
    try {
      return getHostResourceProvider().install(clusterName, hostName, skipInstallForComponents,
        dontSkipInstallForComponents, skipFailure, true);
    } catch (Exception e) {
      LOG.error("INSTALL Host request submission failed:", e);
      throw new RuntimeException("INSTALL Host request submission failed: " + e, e);
    }
  }

  public RequestStatusResponse startHost(String hostName, String clusterName, Collection<String> installOnlyComponents, boolean skipFailure) {
    try {
      return getHostComponentResourceProvider().start(clusterName, hostName, installOnlyComponents, skipFailure, true);
    } catch (Exception e) {
      LOG.error("START Host request submission failed:", e);
      throw new RuntimeException("START Host request submission failed: " + e, e);
    }
  }

  /**
   * Persist cluster state for the ambari UI.  Setting this state informs that UI that a cluster has been
   * installed and started and that the monitoring screen for the cluster should be displayed to the user.
   *
   * @param clusterName  cluster name
   * @param stackName    stack name
   * @param stackVersion stack version
   */
  public void persistInstallStateForUI(String clusterName, String stackName, String stackVersion) {
    String stackInfo = String.format("%s-%s", stackName, stackVersion);
    final ClusterRequest clusterRequest = new ClusterRequest(null, clusterName, "INSTALLED", null, stackInfo, null);

    try {
      RetryHelper.executeWithRetry(new Callable<Object>() {
        @Override
        public Object call() throws Exception {
          getController().updateClusters(Collections.singleton(clusterRequest), null);
          return null;
        }
      });
    } catch (AmbariException e) {
      LOG.error("Unable to set install state for UI", e);
    }
  }

  //todo: non topology type shouldn't be returned
  public List<ConfigurationRequest> createConfigurationRequests(Map<String, Object> clusterProperties) {
    return AbstractResourceProvider.getConfigurationRequests("Clusters", clusterProperties);
  }

  public void setConfigurationOnCluster(final ClusterRequest clusterRequest) {
    try {
      RetryHelper.executeWithRetry(new Callable<Object>() {
        @Override
        public Object call() throws Exception {
          getController().updateClusters(Collections.singleton(clusterRequest), null,
              false, false);
          return null;
        }
      });
    } catch (AmbariException e) {
      LOG.error("Failed to set configurations on cluster: ", e);
      throw new RuntimeException("Failed to set configurations on cluster: " + e, e);
    }
  }

  public void notifyAgentsAboutConfigsChanges(String clusterName) {
    try {
      configHelper.get().updateAgentConfigs(Collections.singleton(clusterName));
    } catch (AmbariException e) {
      LOG.error("Failed to set send agent updates: ", e);
      throw new RuntimeException("Failed to set send agent updates: " + e, e);
    }
  }

  /**
   * Verifies that all desired configurations have reached the resolved state
   *   before proceeding with the install
   *
   * @param clusterName name of the cluster
   * @param updatedConfigTypes set of config types that are required to be in the TOPOLOGY_RESOLVED state
   *
   * @throws AmbariException upon any system-level error that occurs
   */
  public void waitForConfigurationResolution(String clusterName, Set<String> updatedConfigTypes) throws AmbariException {
    Cluster cluster = getController().getClusters().getCluster(clusterName);
    boolean shouldWaitForResolution = true;
    while (shouldWaitForResolution) {
      int numOfRequestsStillRequiringResolution = 0;

      // for all config types specified
      for (String actualConfigType : updatedConfigTypes) {
        // get the actual cluster config for comparison
        DesiredConfig actualConfig = cluster.getDesiredConfigs().get(actualConfigType);
        if (actualConfig == null || actualConfigType.equals("core-site")) {
          continue;
        }
        if (!actualConfig.getTag().equals(TopologyManager.TOPOLOGY_RESOLVED_TAG)) {
          // if any expected config is not resolved, deployment must wait
          LOG.info("Config type " + actualConfigType + " not resolved yet, Blueprint deployment will wait until configuration update is completed");
          numOfRequestsStillRequiringResolution++;
        } else {
          LOG.info("Config type " + actualConfigType + " is resolved in the cluster config.");
        }
      }

      if (numOfRequestsStillRequiringResolution == 0) {
        // all configs are resolved, deployment can continue
        LOG.info("All required configuration types are in the " + TopologyManager.TOPOLOGY_RESOLVED_TAG + " state.  Blueprint deployment can now continue.");
        shouldWaitForResolution = false;
      } else {
        LOG.info("Waiting for " + numOfRequestsStillRequiringResolution + " configuration types to be resolved before Blueprint deployment can continue");

        try {
          // sleep before checking the config again
          Thread.sleep(100);
        } catch (InterruptedException e) {
          LOG.warn("sleep interrupted");
        }
      }
    }
  }

  /**
   * Verifies if the given cluster has at least one desired configuration transitioned through
   * TopologyManager.INITIAL -> .... -> TopologyManager.TOPOLOGY_RESOLVED -> ....
   * @param clusterId the identifier of the cluster to be checked
   * @return true if the cluster
   */
  public boolean isTopologyResolved(long clusterId) {
    boolean isTopologyResolved = false;
    try {
      Cluster cluster = getController().getClusters().getClusterById(clusterId);

      // Check through the various cluster config versions that these transitioned through TopologyManager.INITIAL -> .... -> TopologyManager.TOPOLOGY_RESOLVED -> ....
      Map<String, Set<DesiredConfig>> allDesiredConfigsByType = cluster.getAllDesiredConfigVersions();

      for (String configType: allDesiredConfigsByType.keySet()) {
        Set<DesiredConfig> desiredConfigVersions = allDesiredConfigsByType.get(configType);

        SortedSet<DesiredConfig> desiredConfigsOrderedByVersion = new TreeSet<>(new Comparator<DesiredConfig>() {
          @Override
          public int compare(DesiredConfig o1, DesiredConfig o2) {
            if (o1.getVersion() < o2.getVersion()) {
              return -1;
            }

            if (o1.getVersion() > o2.getVersion()) {
              return 1;
            }

            return 0;
          }
        });

        desiredConfigsOrderedByVersion.addAll(desiredConfigVersions);

        int tagMatchState = 0; // 0 -> INITIAL -> tagMatchState = 1 -> TOPLOGY_RESOLVED -> tagMatchState = 2

        for (DesiredConfig config: desiredConfigsOrderedByVersion) {
          if (config.getTag().equals(TopologyManager.INITIAL_CONFIG_TAG) && tagMatchState == 0) {
            tagMatchState = 1;
          } else if (config.getTag().equals(TopologyManager.TOPOLOGY_RESOLVED_TAG) && tagMatchState == 1) {
            tagMatchState = 2;
            break;
          }
        }

        if (tagMatchState == 2) {
          isTopologyResolved = true;
          break;
        }
      }

    } catch (ClusterNotFoundException e) {
      LOG.info("Attempted to determine if configuration is topology resolved for a non-existent cluster: {}",
              clusterId);
    } catch (AmbariException e) {
      throw new RuntimeException(
              "Unable to determine if cluster config is topology resolved due to unknown error: " + e, e);
    }

    return isTopologyResolved;
  }

  public PersistedState getPersistedTopologyState() {
    return persistedState;
  }

  public boolean isHostRegisteredWithCluster(long clusterId, String host) {
    boolean found = false;
    try {
      Collection<Host> hosts = getController().getClusters().getClusterById(clusterId).getHosts();
      for (Host h : hosts) {
        if (h.getHostName().equals(host)) {
          found = true;
          break;
        }
      }
    } catch (AmbariException e) {
      throw new RuntimeException(String.format("Unable to get hosts for cluster ID = %s: %s", clusterId, e), e);
    }
    return found;
  }

  public long getClusterId(String clusterName) throws AmbariException {
    return getController().getClusters().getCluster(clusterName).getClusterId();
  }

  public String getClusterName(long clusterId) throws AmbariException {
    return getController().getClusters().getClusterById(clusterId).getClusterName();
  }

  /**
   * Add the new host to an existing config group.
   */
  private boolean addHostToExistingConfigGroups(String hostName, ClusterTopology topology, String configGroupName) {
    boolean addedHost = false;
    Clusters clusters;
    Cluster cluster;
    try {
      clusters = getController().getClusters();
      cluster = clusters.getClusterById(topology.getClusterId());
    } catch (AmbariException e) {
      throw new RuntimeException(String.format(
          "Attempt to add hosts to a non-existent cluster: '%s'", topology.getClusterId()));
    }
    // I don't know of a method to get config group by name
    //todo: add a method to get config group by name
    Map<Long, ConfigGroup> configGroups = cluster.getConfigGroups();
    for (ConfigGroup group : configGroups.values()) {
      if (group.getName().equals(configGroupName)) {
        try {
          Host host = clusters.getHost(hostName);
          addedHost = true;
          if (! group.getHosts().containsKey(host.getHostId())) {
            group.addHost(host);
          }

        } catch (AmbariException e) {
          // shouldn't occur, this host was just added to the cluster
          throw new RuntimeException(String.format(
              "An error occurred while registering host '%s' with config group '%s' ", hostName, group.getName()), e);
        }
      }
    }
    return addedHost;
  }

  /**
   * Register config groups for host group scoped configuration.
   * For each host group with configuration specified in the blueprint, a config group is created
   * and the hosts associated with the host group are assigned to the config group.
   */
  private void createConfigGroupsAndRegisterHost(ClusterTopology topology, String groupName) throws AmbariException {
    Map<String, Map<String, Config>> groupConfigs = new HashMap<>();
    Stack stack = topology.getBlueprint().getStack();

    // get the host-group config with cluster creation template overrides
    Configuration topologyHostGroupConfig = topology.
        getHostGroupInfo().get(groupName).getConfiguration();

    // only get user provided configuration for host group which includes only CCT/HG and BP/HG properties
    Map<String, Map<String, String>> userProvidedGroupProperties =
        topologyHostGroupConfig.getFullProperties(1);

    // iterate over topo host group configs which were defined in
    for (Map.Entry<String, Map<String, String>> entry : userProvidedGroupProperties.entrySet()) {
      String type = entry.getKey();
      List<String> services = stack.getServicesForConfigType(type);
      String service = services.stream()
        .filter(each -> topology.getBlueprint().getServices().contains(each))
        .findFirst()
        .orElseThrow(() -> new IllegalArgumentException("Specified configuration type is not associated with any service: " + type));

      Config config = configFactory.createReadOnly(type, groupName, entry.getValue(), null);
      //todo: attributes
      Map<String, Config> serviceConfigs = groupConfigs.get(service);
      if (serviceConfigs == null) {
        serviceConfigs = new HashMap<>();
        groupConfigs.put(service, serviceConfigs);
      }
      serviceConfigs.put(type, config);
    }

    String bpName = topology.getBlueprint().getName();
    for (Map.Entry<String, Map<String, Config>> entry : groupConfigs.entrySet()) {
      String service = entry.getKey();
      Map<String, Config> serviceConfigs = entry.getValue();
      String absoluteGroupName = getConfigurationGroupName(bpName, groupName);
      Collection<String> groupHosts;

      groupHosts = topology.getHostGroupInfo().
          get(groupName).getHostNames();

      // remove hosts that are not assigned to the cluster yet
      String clusterName = null;
      try {
        clusterName = getClusterName(topology.getClusterId());
      } catch (AmbariException e) {
        LOG.error("Cannot get cluster name for clusterId = " + topology.getClusterId(), e);
        throw new RuntimeException(e);
      }

      final Map<String, Host> clusterHosts = getController().getClusters().getHostsForCluster(clusterName);
      Iterable<String> filteredGroupHosts = Iterables.filter(groupHosts, new com.google.common.base.Predicate<String>() {
        @Override
        public boolean apply(@Nullable String groupHost) {
          return clusterHosts.containsKey(groupHost);
        }
      });

      ConfigGroupRequest request = new ConfigGroupRequest(null, clusterName,
        absoluteGroupName, service, service, "Host Group Configuration",
        Sets.newHashSet(filteredGroupHosts), serviceConfigs);

      // get the config group provider and create config group resource
      ConfigGroupResourceProvider configGroupProvider = (ConfigGroupResourceProvider)
          getClusterController().ensureResourceProvider(Resource.Type.ConfigGroup);

      try {
        configGroupProvider.createResources(Collections.singleton(request));
      } catch (Exception e) {
        LOG.error("Failed to create new configuration group: " + e);
        throw new RuntimeException("Failed to create new configuration group: " + e, e);
      }
    }
  }

  /**
   * Get a config group name based on a bp and host group.
   *
   * @param bpName        blueprint name
   * @param hostGroupName host group name
   * @return config group name
   */
  private String getConfigurationGroupName(String bpName, String hostGroupName) {
    return String.format("%s:%s", bpName, hostGroupName);
  }

  /**
   * Gets an instance of {@link ConfigHelper} for classes which are not
   * dependency injected.
   *
   * @return a {@link ConfigHelper} instance.
   */
  public ConfigHelper getConfigHelper() {
    return configHelper.get();
  }

  private synchronized HostResourceProvider getHostResourceProvider() {
    if (hostResourceProvider == null) {
      hostResourceProvider = (HostResourceProvider)
          ClusterControllerHelper.getClusterController().ensureResourceProvider(Resource.Type.Host);

    }
    return hostResourceProvider;
  }

  private synchronized HostComponentResourceProvider getHostComponentResourceProvider() {
    if (hostComponentResourceProvider == null) {
      hostComponentResourceProvider = (HostComponentResourceProvider)
          ClusterControllerHelper.getClusterController().ensureResourceProvider(Resource.Type.HostComponent);

    }
    return hostComponentResourceProvider;
  }

  private synchronized ServiceResourceProvider getServiceResourceProvider() {
    if (serviceResourceProvider == null) {
      serviceResourceProvider = (ServiceResourceProvider) ClusterControllerHelper.
          getClusterController().ensureResourceProvider(Resource.Type.Service);
    }
    return serviceResourceProvider;
  }

  private synchronized ComponentResourceProvider getComponentResourceProvider() {
    if (componentResourceProvider == null) {
      componentResourceProvider = (ComponentResourceProvider) ClusterControllerHelper.
          getClusterController().ensureResourceProvider(Resource.Type.Component);
    }
    return componentResourceProvider;
  }

  private synchronized VersionDefinitionResourceProvider getVersionDefinitionResourceProvider() {
    if (versionDefinitionResourceProvider == null) {
      versionDefinitionResourceProvider = (VersionDefinitionResourceProvider) ClusterControllerHelper.
          getClusterController().ensureResourceProvider(Resource.Type.VersionDefinition);
    }
    return versionDefinitionResourceProvider;

  }

}
