/*
 * 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.state;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.TimeUnit;

import javax.annotation.Nullable;

import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.AmbariRuntimeException;
import org.apache.ambari.server.HostNotFoundException;
import org.apache.ambari.server.agent.stomp.AgentConfigsHolder;
import org.apache.ambari.server.agent.stomp.MetadataHolder;
import org.apache.ambari.server.agent.stomp.dto.ClusterConfigs;
import org.apache.ambari.server.api.services.AmbariMetaInfo;
import org.apache.ambari.server.configuration.Configuration;
import org.apache.ambari.server.controller.AmbariManagementController;
import org.apache.ambari.server.controller.AmbariManagementControllerImpl;
import org.apache.ambari.server.events.AgentConfigsUpdateEvent;
import org.apache.ambari.server.events.HostComponentUpdate;
import org.apache.ambari.server.events.HostComponentsUpdateEvent;
import org.apache.ambari.server.events.publishers.STOMPUpdatePublisher;
import org.apache.ambari.server.orm.dao.ClusterDAO;
import org.apache.ambari.server.orm.dao.ServiceConfigDAO;
import org.apache.ambari.server.orm.entities.ClusterConfigEntity;
import org.apache.ambari.server.orm.entities.HostComponentDesiredStateEntity;
import org.apache.ambari.server.orm.entities.ServiceConfigEntity;
import org.apache.ambari.server.stack.StackDirectory;
import org.apache.ambari.server.state.PropertyInfo.PropertyType;
import org.apache.ambari.server.state.configgroup.ConfigGroup;
import org.apache.ambari.server.utils.SecretReference;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.math.NumberUtils;
import org.apache.commons.lang3.StringEscapeUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Objects;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.collect.Maps;
import com.google.inject.Inject;
import com.google.inject.Provider;
import com.google.inject.Singleton;
import com.google.inject.persist.Transactional;

/**
 * Helper class that works with config traversals.
 */
@Singleton
public class ConfigHelper {

  private Clusters clusters = null;
  private AmbariMetaInfo ambariMetaInfo = null;
  private ClusterDAO clusterDAO = null;
  private static final String DELETED = "DELETED_";
  public static final String CLUSTER_DEFAULT_TAG = "tag";
  private final boolean STALE_CONFIGS_CACHE_ENABLED;
  private final int STALE_CONFIGS_CACHE_EXPIRATION_TIME;

  /**
   * Cache for storing stale config flags. Key for cache is hash of [actualConfigs, desiredConfigs, hostName, serviceName,
   * componentName].
   */
  private final Cache<Integer, Boolean> staleConfigsCache;

  /**
   * clusterId -> hostId -> serviceName -> serviceComponentName -> state map to reduce redundant updates sending.
   */
  private final Map<Long, Map<Long, Map<String, Map<String, Boolean>>>> stateCache = new HashMap<>();

  private final Cache<Integer, String> refreshConfigCommandCache;

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

  /**
   * List of property prefixes and names. Please keep in alphabetical order.
   */
  public static final String HBASE_SITE = "hbase-site";
  public static final String HDFS_SITE = "hdfs-site";
  public static final String HIVE_SITE = "hive-site";
  public static final String YARN_SITE = "yarn-site";
  public static final String CLUSTER_ENV = "cluster-env";
  public static final String CLUSTER_ENV_ALERT_REPEAT_TOLERANCE = "alerts_repeat_tolerance";
  public static final String CLUSTER_ENV_RETRY_ENABLED = "command_retry_enabled";
  public static final String SERVICE_CHECK_TYPE = "service_check_type";
  public static final String CLUSTER_ENV_RETRY_COMMANDS = "commands_to_retry";
  public static final String CLUSTER_ENV_RETRY_MAX_TIME_IN_SEC = "command_retry_max_time_in_sec";
  public static final String COMMAND_RETRY_MAX_TIME_IN_SEC_DEFAULT = "600";
  public static final String CLUSTER_ENV_STACK_NAME_PROPERTY = "stack_name";
  public static final String CLUSTER_ENV_STACK_FEATURES_PROPERTY = "stack_features";
  public static final String CLUSTER_ENV_STACK_TOOLS_PROPERTY = "stack_tools";
  public static final String CLUSTER_ENV_STACK_ROOT_PROPERTY = "stack_root";
  public static final String CLUSTER_ENV_STACK_PACKAGES_PROPERTY = "stack_packages";

  public static final String HTTP_ONLY = "HTTP_ONLY";
  public static final String HTTPS_ONLY = "HTTPS_ONLY";
  public static final String SERVICE_CHECK_MINIMAL = "minimal";

  /**
   * The tag given to newly created versions.
   */
  public static final String FIRST_VERSION_TAG = "version1";

  @Inject
  private Provider<MetadataHolder> m_metadataHolder;

  @Inject
  private Provider<AgentConfigsHolder> m_agentConfigsHolder;

  @Inject
  private Provider<AmbariManagementControllerImpl> m_ambariManagementController;

  @Inject
  private STOMPUpdatePublisher STOMPUpdatePublisher;

  @Inject
  private ServiceConfigDAO serviceConfigDAO;

  @Inject
  public ConfigHelper(Clusters c, AmbariMetaInfo metaInfo, Configuration configuration, ClusterDAO clusterDAO) {
    clusters = c;
    ambariMetaInfo = metaInfo;
    this.clusterDAO = clusterDAO;
    STALE_CONFIGS_CACHE_ENABLED = configuration.isStaleConfigCacheEnabled();
    STALE_CONFIGS_CACHE_EXPIRATION_TIME = configuration.staleConfigCacheExpiration();
    staleConfigsCache = CacheBuilder.newBuilder().
        expireAfterWrite(STALE_CONFIGS_CACHE_EXPIRATION_TIME, TimeUnit.SECONDS).build();

    refreshConfigCommandCache = CacheBuilder.newBuilder().
            expireAfterWrite(STALE_CONFIGS_CACHE_EXPIRATION_TIME, TimeUnit.SECONDS).build();
  }

  /**
   * Gets the desired tags for a cluster and host
   *
   * @param cluster  the cluster
   * @param hostName the host name
   * @return a map of tag type to tag names with overrides
   * @throws AmbariException
   */
  public Map<String, Map<String, String>> getEffectiveDesiredTags(
          Cluster cluster, String hostName) throws AmbariException {

    return getEffectiveDesiredTags(cluster, hostName, null);
  }

  /**
   * Gets the desired tags for a cluster and host
   *
   * @param cluster
   *          the cluster
   * @param hostName
   *          the host name
   * @return a map of tag type to tag names with overrides
   * @throws AmbariException
   */
  public Map<String, Map<String, String>> getEffectiveDesiredTags(Cluster cluster, String hostName,
                                                                  @Nullable Map<String, DesiredConfig> desiredConfigs) throws AmbariException {
    Host host = null;
    if (hostName != null) {
      try {
        host = clusters.getHost(hostName);
      } catch (HostNotFoundException e) {
        LOG.error("Cannot get desired config for unknown host {}", hostName, e);
      }
    }
    Map<String, HostConfig> desiredHostConfigs = (host == null) ? null
        : host.getDesiredHostConfigs(cluster, desiredConfigs);

    return getEffectiveDesiredTags(cluster, desiredConfigs, desiredHostConfigs);
  }

  /**
   * Gets the desired tags for a cluster and overrides for a host
   *
   * @param cluster
   *          the cluster
   * @param hostConfigOverrides
   *          the host overrides applied using config groups
   * @param clusterDesired
   *          the desired configurations for the cluster. Obtaining these can be
   *          expensive, ans since this method could be called 10,000's of times
   *          when generating cluster/host responses. Therefore, the caller
   *          should build these once and pass them in. If {@code null}, then
   *          this method will retrieve them at runtime, incurring a performance
   *          penality.
   * @return a map of tag type to tag names with overrides
   */
  private Map<String, Map<String, String>> getEffectiveDesiredTags(Cluster cluster, Map<String,
          DesiredConfig> clusterDesired, Map<String, HostConfig> hostConfigOverrides) {

    if (null == cluster) {
      clusterDesired = new HashMap<>();
    }

    // per method contract, lookup if not supplied
    if (null == clusterDesired) {
      clusterDesired = cluster.getDesiredConfigs();
    }

    if (null == clusterDesired) {
      clusterDesired = new HashMap<>();
    }

    Map<String, Map<String, String>> resolved = new TreeMap<>();

    // Do not use host component config mappings.  Instead, the rules are:
    // 1) Use the cluster desired config
    // 2) override (1) with config-group overrides

    for (Entry<String, DesiredConfig> clusterEntry : clusterDesired.entrySet()) {
      String type = clusterEntry.getKey();
      String tag = clusterEntry.getValue().getTag();

      // 1) start with cluster config
      if (cluster != null) {
        Config config = cluster.getConfig(type, tag);
        if (null == config) {
          continue;
        }

        Map<String, String> tags = new LinkedHashMap<>();

        tags.put(CLUSTER_DEFAULT_TAG, config.getTag());

        // AMBARI-3672. Only consider Config groups for override tags
        // tags -> (configGroupId, versionTag)
        if (hostConfigOverrides != null) {
          HostConfig hostConfig = hostConfigOverrides.get(config.getType());
          if (hostConfig != null) {
            for (Entry<Long, String> tagEntry : hostConfig
                    .getConfigGroupOverrides().entrySet()) {
              tags.put(tagEntry.getKey().toString(), tagEntry.getValue());
            }
          }
        }

        resolved.put(type, tags);
      }
    }

    return resolved;
  }


  public Set<String> filterInvalidPropertyValues(Map<PropertyInfo, String> properties, String filteredListName) {
    Set<String> resultSet = new HashSet<>();
    for (Iterator<Entry<PropertyInfo, String>> iterator = properties.entrySet().iterator(); iterator.hasNext();) {
      Entry<PropertyInfo, String> property = iterator.next();
      PropertyInfo propertyInfo = property.getKey();
      String propertyValue = property.getValue();
      if (propertyValue == null || propertyValue.toLowerCase().equals("null") || propertyValue.isEmpty()) {
        LOG.error(String.format("Excluding property %s from %s, because of invalid or empty value!", propertyInfo.getName(), filteredListName));
        iterator.remove();
      } else {
        resultSet.add(propertyValue);
      }
    }
    return resultSet;
  }

  /**
   * Get all config properties for a cluster given a set of configType to
   * versionTags map. This helper method merges all the override tags with a
   * the properties from parent cluster config properties
   *
   * @param cluster
   * @param desiredTags
   * @return {type : {key, value}}
   */
  public Map<String, Map<String, String>> getEffectiveConfigProperties(
      Cluster cluster, Map<String, Map<String, String>> desiredTags) {

    Map<String, Map<String, String>> properties = new HashMap<>();

    if (desiredTags != null) {
      for (Entry<String, Map<String, String>> entry : desiredTags.entrySet()) {
        String type = entry.getKey();
        Map<String, String> propertyMap = properties.get(type);
        if (propertyMap == null) {
          propertyMap = new HashMap<>();
        }

        Map<String, String> tags = new HashMap<>(entry.getValue());
        String clusterTag = tags.get(CLUSTER_DEFAULT_TAG);

        // Overrides is only supported if the config type exists at cluster
        // level
        if (clusterTag != null) {
          Config config = cluster.getConfig(type, clusterTag);
          if (config != null) {
            propertyMap.putAll(config.getProperties());
          }
          tags.remove(CLUSTER_DEFAULT_TAG);
          // Now merge overrides
          for (Entry<String, String> overrideEntry : tags.entrySet()) {
            Config overrideConfig = cluster.getConfig(type,
                overrideEntry.getValue());

            if (overrideConfig != null) {
              propertyMap = getMergedConfig(propertyMap, overrideConfig.getProperties());
            }
          }
        }
        properties.put(type, propertyMap);
      }
    }

    return properties;
  }

  public Map<String, Map<String, String>> getEffectiveConfigProperties(String clusterName, String hostName) throws AmbariException {
    Cluster cluster = clusters.getCluster(clusterName);
    return getEffectiveConfigProperties(cluster, getEffectiveDesiredTags(cluster, hostName));
  }

  /**
   * Get all config attributes for a cluster given a set of configType to
   * versionTags map. This helper method merges all the override tags with a
   * the attributes from parent cluster config properties
   *
   * @param cluster
   * @param desiredTags
   * @return {type : {attribute : {property, attributeValue}}
   */
  public Map<String, Map<String, Map<String, String>>> getEffectiveConfigAttributes(
      Cluster cluster, Map<String, Map<String, String>> desiredTags) {

    Map<String, Map<String, Map<String, String>>> attributes = new HashMap<>();

    if (desiredTags != null) {
      for (Entry<String, Map<String, String>> entry : desiredTags.entrySet()) {

        String type = entry.getKey();
        Map<String, Map<String, String>> attributesMap = null;

        Map<String, String> tags = new HashMap<>(entry.getValue());
        String clusterTag = tags.get(CLUSTER_DEFAULT_TAG);

        if (clusterTag != null) {
          Config config = cluster.getConfig(type, clusterTag);
          if (config != null) {
            attributesMap = new TreeMap<>();
            cloneAttributesMap(config.getPropertiesAttributes(), attributesMap);
          }
          tags.remove(CLUSTER_DEFAULT_TAG);
        }
        for (Entry<String, String> overrideEntry : tags.entrySet()) {
          Config overrideConfig = cluster.getConfig(type,
              overrideEntry.getValue());
          overrideAttributes(overrideConfig, attributesMap);
        }
        if (attributesMap != null) {
          attributes.put(type, attributesMap);
        }
      }
    }

    return attributes;
  }

  /**
   * Merge override with original, if original property doesn't exist,
   * add it to the properties
   *
   * @param persistedClusterConfig
   * @param override
   * @return
   */
  public Map<String, String> getMergedConfig(Map<String,
      String> persistedClusterConfig, Map<String, String> override) {

    Map<String, String> finalConfig = new HashMap<>(persistedClusterConfig);

    if (override != null && override.size() > 0) {
      for (Entry<String, String> entry : override.entrySet()) {
        Boolean deleted = 0 == entry.getKey().indexOf(DELETED);
        String nameToUse = deleted ?
            entry.getKey().substring(DELETED.length()) : entry.getKey();
        if (finalConfig.containsKey(nameToUse)) {
          finalConfig.remove(nameToUse);
        }
        if (!deleted) {
          finalConfig.put(nameToUse, entry.getValue());
        }
      }
    }

    return finalConfig;
  }

  /**
   * Retrieves effective configurations for specified cluster and tags and merge them with
   * present before in {@code configurations}
   * @param configurations configurations will be merged with effective cluster configurations
   * @param configurationTags configuration tags for cluster's desired configs
   * @param cluster cluster to configs retrieving
   */
  public void getAndMergeHostConfigs(Map<String, Map<String, String>> configurations,
                                     Map<String, Map<String, String>> configurationTags,
                                     Cluster cluster) {
    if (null != configurationTags && !configurationTags.isEmpty()) {
      Map<String, Map<String, String>> configProperties =
          getEffectiveConfigProperties(cluster, configurationTags);

      // Apply the configurations present before on top of derived configs
      for (Map.Entry<String, Map<String, String>> entry : configProperties.entrySet()) {
        String type = entry.getKey();
        Map<String, String> allLevelMergedConfig = entry.getValue();

        if (configurations.containsKey(type)) {
          Map<String, String> mergedConfig = getMergedConfig(allLevelMergedConfig,
              configurations.get(type));

          configurations.get(type).clear();
          configurations.get(type).putAll(mergedConfig);

        } else {
          configurations.put(type, new HashMap<>());
          configurations.get(type).putAll(allLevelMergedConfig);
        }
      }
    }
  }

  /**
   * Retrieves effective configuration attributes for specified cluster and tags and merge them with
   * present before in {@code configurationAttributes}
   * @param configurationAttributes configuration attributes will be merged with effective ones on the cluster
   * @param configurationTags  configuration tags for cluster's desired configs
   * @param cluster cluster to config attributes retrieving
   */
  public void getAndMergeHostConfigAttributes(Map<String, Map<String, Map<String, String>>> configurationAttributes,
                                     Map<String, Map<String, String>> configurationTags,
                                     Cluster cluster) {
    if (null != configurationTags && !configurationTags.isEmpty()) {
      Map<String, Map<String, Map<String, String>>> configAttributes =
          getEffectiveConfigAttributes(cluster, configurationTags);

      for (Map.Entry<String, Map<String, Map<String, String>>> attributesOccurrence : configAttributes.entrySet()) {
        String type = attributesOccurrence.getKey();
        Map<String, Map<String, String>> attributes = attributesOccurrence.getValue();

        if (configurationAttributes != null) {
          if (!configurationAttributes.containsKey(type)) {
            configurationAttributes.put(type,
                new TreeMap<>());
          }
          cloneAttributesMap(attributes,
              configurationAttributes.get(type));
        }
      }
    }
  }

  /**
   * Merge override attributes with original ones.
   * If overrideConfig#getPropertiesAttributes does not contain occurrence of override for any of
   * properties from overrideConfig#getProperties then persisted attribute should be removed.
   */
  public Map<String, Map<String, String>> overrideAttributes(Config overrideConfig,
                                                             Map<String, Map<String, String>> persistedAttributes) {
    if (overrideConfig != null && persistedAttributes != null) {
      Map<String, Map<String, String>> overrideAttributes = overrideConfig.getPropertiesAttributes();
      if (overrideAttributes != null) {
        cloneAttributesMap(overrideAttributes, persistedAttributes);
        Map<String, String> overrideProperties = overrideConfig.getProperties();
        if (overrideProperties != null) {
          Set<String> overriddenProperties = overrideProperties.keySet();
          for (String overriddenProperty : overriddenProperties) {
            for (Entry<String, Map<String, String>> persistedAttribute : persistedAttributes.entrySet()) {
              String attributeName = persistedAttribute.getKey();
              Map<String, String> persistedAttributeValues = persistedAttribute.getValue();
              Map<String, String> overrideAttributeValues = overrideAttributes.get(attributeName);
              if (overrideAttributeValues == null || !overrideAttributeValues.containsKey(overriddenProperty)) {
                persistedAttributeValues.remove(overriddenProperty);
              }
            }
          }
        }
      }
    }
    return persistedAttributes;
  }

  public void cloneAttributesMap(Map<String, Map<String, String>> sourceAttributesMap,
                                 Map<String, Map<String, String>> targetAttributesMap) {
    if (sourceAttributesMap != null && targetAttributesMap != null) {
      for (Entry<String, Map<String, String>> attributesEntry : sourceAttributesMap.entrySet()) {
        String attributeName = attributesEntry.getKey();
        if (!targetAttributesMap.containsKey(attributeName)) {
          targetAttributesMap.put(attributeName, new TreeMap<>());
        }
        for (Entry<String, String> attributesValue : attributesEntry.getValue().entrySet()) {
          targetAttributesMap.get(attributeName).put(attributesValue.getKey(), attributesValue.getValue());
        }
      }
    }
  }

  public void applyCustomConfig(Map<String, Map<String, String>> configurations,
                                String type, String name, String value, Boolean deleted) {
    if (!configurations.containsKey(type)) {
      configurations.put(type, new HashMap<>());
    }
    String nameToUse = deleted ? DELETED + name : name;
    Map<String, String> properties = configurations.get(type);
    if (properties.containsKey(nameToUse)) {
      properties.remove(nameToUse);
    }
    properties.put(nameToUse, value);
  }

  /**
   * The purpose of this method is to determine if a
   * {@link ServiceComponentHost}'s known actual configs are different than what
   * is set on the cluster (the desired). The following logic is applied:
   * <ul>
   * <li>Desired type does not exist on the SCH (actual)
   * <ul>
   * <li>Type does not exist on the stack: <code>false</code></li>
   * <li>Type exists on the stack: <code>true</code> if the config key is on the
   * stack. otherwise <code>false</code></li>
   * </ul>
   * </li>
   * <li>Desired type exists for the SCH
   * <ul>
   * <li>Desired tags already set for the SCH (actual): <code>false</code></li>
   * <li>Desired tags DO NOT match SCH: <code>true</code> if the changed keys
   * exist on the stack, otherwise <code>false</code></li>
   * </ul>
   * </li>
   * </ul>
   *
   * @param sch
   *          the SCH to calcualte config staleness for (not {@code null}).
   * @param requestDesiredConfigs
   *          the desired configurations for the cluster. Obtaining these can be
   *          expensive and since this method operates on SCH's, it could be
   *          called 10,000's of times when generating cluster/host responses.
   *          Therefore, the caller should build these once and pass them in. If
   *          {@code null}, then this method will retrieve them at runtime,
   *          incurring a performance penality.
   *
   * @return <code>true</code> if the actual configs are stale
   */
  public boolean isStaleConfigs(ServiceComponentHost sch, Map<String, DesiredConfig> requestDesiredConfigs)
      throws AmbariException {
    HostComponentDesiredStateEntity hostComponentDesiredStateEntity = sch.getDesiredStateEntity();
    return isStaleConfigs(sch, requestDesiredConfigs, hostComponentDesiredStateEntity);
  }

  public boolean isStaleConfigs(ServiceComponentHost sch, Map<String, DesiredConfig> requestDesiredConfigs,
                                HostComponentDesiredStateEntity hostComponentDesiredStateEntity)
          throws AmbariException {
    boolean stale = calculateIsStaleConfigs(sch, requestDesiredConfigs, hostComponentDesiredStateEntity);
    if (LOG.isDebugEnabled()) {
      LOG.debug("Cache configuration staleness for host {} and component {} as {}",
              sch.getHostName(), sch.getServiceComponentName(), stale);
    }
    return stale;
  }

  /**
   * Remove configs by type
   *
   * @param type config Type
   */
  @Transactional
  public void removeConfigsByType(Cluster cluster, String type) {
    Set<String> globalVersions = cluster.getConfigsByType(type).keySet();

    for (String version : globalVersions) {
      ClusterConfigEntity clusterConfigEntity = clusterDAO.findConfig
          (cluster.getClusterId(), type, version);

      clusterDAO.removeConfig(clusterConfigEntity);
    }
  }

  /**
   * Gets all the config dictionary where property with the given name is present in stack definitions
   *
   * @param stackId
   * @param propertyName
   */
  public Set<String> findConfigTypesByPropertyName(StackId stackId, String propertyName, String clusterName) throws AmbariException {
    StackInfo stack = ambariMetaInfo.getStack(stackId.getStackName(),
                                              stackId.getStackVersion());

    Set<String> result = new HashSet<>();

    for (Service service : clusters.getCluster(clusterName).getServices().values()) {
      Set<PropertyInfo> stackProperties = ambariMetaInfo.getServiceProperties(stack.getName(), stack.getVersion(), service.getName());
      Set<PropertyInfo> stackLevelProperties = ambariMetaInfo.getStackProperties(stack.getName(), stack.getVersion());
      stackProperties.addAll(stackLevelProperties);

      for (PropertyInfo stackProperty : stackProperties) {
        if (stackProperty.getName().equals(propertyName)) {
          String configType = fileNameToConfigType(stackProperty.getFilename());

          result.add(configType);
        }
      }
    }

    return result;
  }

  /**
   * Gets a map of config types to password property names to password property value names,
   * that are credential store enabled.
   *
   * @param stackId
   * @param service
   * @return
   * @throws AmbariException
     */
  public Map<String, Map<String, String>> getCredentialStoreEnabledProperties(StackId stackId, Service service)
          throws AmbariException {
    PropertyType propertyType = PropertyType.PASSWORD;
    StackInfo stack = ambariMetaInfo.getStack(stackId.getStackName(), stackId.getStackVersion());
    Map<String, Map<String, String>> result = new HashMap<>();
    Map<String, String> passwordProperties;
    Set<PropertyInfo> serviceProperties = ambariMetaInfo.getServiceProperties(stack.getName(), stack.getVersion(), service.getName());
    for (PropertyInfo serviceProperty : serviceProperties) {
      if (serviceProperty.getPropertyTypes().contains(propertyType)) {
        if (!serviceProperty.getPropertyValueAttributes().isKeyStore()) {
          continue;
        }
        String stackPropertyConfigType = fileNameToConfigType(serviceProperty.getFilename());
        passwordProperties = result.get(stackPropertyConfigType);
        if (passwordProperties == null) {
          passwordProperties = new HashMap<>();
          result.put(stackPropertyConfigType, passwordProperties);
        }
        // If the password property is used by another property, it means the password property
        // is a password value name while the use is the password alias name. If the user property
        // is from another config type, include that in the password alias name as name:type.
        if (serviceProperty.getUsedByProperties().size() > 0) {
          for (PropertyDependencyInfo usedByProperty : serviceProperty.getUsedByProperties()) {
            String propertyName = usedByProperty.getName();
            if (!StringUtils.isEmpty(usedByProperty.getType())) {
              propertyName += ':' + usedByProperty.getType();
            }
            passwordProperties.put(propertyName, serviceProperty.getName());
          }
        }
        else {
          passwordProperties.put(serviceProperty.getName(), serviceProperty.getName());
        }
      }
    }

    return result;
  }

  /***
   * Fetch user to group mapping from the cluster configs. UserGroupEntries contain information regarding the group that the user is associated to.
   * @param stackId
   * @param cluster
   * @param desiredConfigs
   * @return
   * @throws AmbariException
   */
  public Map<String, Set<String>> createUserGroupsMap(StackId stackId,
                                                      Cluster cluster, Map<String, DesiredConfig> desiredConfigs) throws AmbariException {
    StackInfo stack = ambariMetaInfo.getStack(stackId.getStackName(), stackId.getStackVersion());
    Map<String, ServiceInfo> servicesMap = ambariMetaInfo.getServices(stack.getName(), stack.getVersion());
    Set<PropertyInfo> stackProperties = ambariMetaInfo.getStackProperties(stack.getName(), stack.getVersion());
    return createUserGroupsMap(cluster, desiredConfigs, servicesMap, stackProperties);
  }

  /***
   * Fetch user to group mapping from the cluster configs. UserGroupEntries contain information regarding the group that the user is associated to.
   * @param cluster
   * @param desiredConfigs
   * @param servicesMap
   * @param stackProperties
   * @return
   * @throws AmbariException
   */
  public Map<String, Set<String>> createUserGroupsMap(
    Cluster cluster, Map<String, DesiredConfig> desiredConfigs,
    Map<String, ServiceInfo> servicesMap, Set<PropertyInfo> stackProperties) throws AmbariException {

    Map<String, Set<String>> userGroupsMap = new HashMap<>();
    Map<PropertyInfo, String> userProperties = getPropertiesWithPropertyType(
      PropertyType.USER, cluster, desiredConfigs, servicesMap, stackProperties);
    Map<PropertyInfo, String> groupProperties = getPropertiesWithPropertyType(
      PropertyType.GROUP, cluster, desiredConfigs, servicesMap, stackProperties);

    if(userProperties != null && groupProperties != null) {
      for(Map.Entry<PropertyInfo, String> userProperty : userProperties.entrySet()) {
        PropertyInfo userPropertyInfo = userProperty.getKey();
        String userPropertyValue = userProperty.getValue();
        if(userPropertyInfo.getPropertyValueAttributes() != null
          && userPropertyInfo.getPropertyValueAttributes().getUserGroupEntries() != null) {
          Set<String> groupPropertyValues = new HashSet<>();
          Collection<UserGroupInfo> userGroupEntries = userPropertyInfo.getPropertyValueAttributes().getUserGroupEntries();
          for (UserGroupInfo userGroupInfo : userGroupEntries) {
            boolean found = false;
            for(Map.Entry<PropertyInfo, String> groupProperty : groupProperties.entrySet()) {
              PropertyInfo groupPropertyInfo = groupProperty.getKey();
              String groupPropertyValue = groupProperty.getValue();
              if(StringUtils.equals(userGroupInfo.getType(),
                ConfigHelper.fileNameToConfigType(groupPropertyInfo.getFilename()))
                && StringUtils.equals(userGroupInfo.getName(), groupPropertyInfo.getName())) {
                groupPropertyValues.add(groupPropertyValue);
                found = true;
              }
            }

            if(!found) {
              //Log error if the user-group mapping is not found
              LOG.error("User group mapping property {" + userGroupInfo.getType() + "/" + userGroupInfo.getName() + "} is missing for user property {" + ConfigHelper.fileNameToConfigType(userPropertyInfo.getFilename()) + "/" + userPropertyInfo.getName() + "} (username = " + userPropertyInfo.getValue() +")");
            }
          }
          userGroupsMap.put(userPropertyValue, groupPropertyValues);
        }
      }
    }
    return userGroupsMap;
  }

  /***
   * Fetch all the properties of a given PropertyType. For eg: Fetch all cluster configs that are of type "user"
   * @param stackId
   * @param propertyType
   * @param cluster
   * @param desiredConfigs
   * @return
   * @throws AmbariException
   */
  public Map<PropertyInfo, String> getPropertiesWithPropertyType(StackId stackId, PropertyType propertyType,
                                                                 Cluster cluster, Map<String, DesiredConfig> desiredConfigs) throws AmbariException {
    StackInfo stack = ambariMetaInfo.getStack(stackId.getStackName(), stackId.getStackVersion());
