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

import static java.util.stream.Collectors.toSet;
import static org.apache.ambari.server.controller.internal.HostComponentResourceProvider.ALL_COMPONENTS;
import static org.apache.ambari.server.controller.internal.ProvisionAction.INSTALL_AND_START;
import static org.apache.ambari.server.controller.internal.ProvisionAction.INSTALL_ONLY;
import static org.apache.ambari.server.state.ServiceInfo.HADOOP_COMPATIBLE_FS;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.controller.RequestStatusResponse;
import org.apache.ambari.server.controller.internal.ProvisionAction;
import org.apache.ambari.server.controller.internal.ProvisionClusterRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Represents a cluster topology.
 * Topology includes the the associated blueprint, cluster configuration and hostgroup -> host mapping.
 */
public class ClusterTopologyImpl implements ClusterTopology {

  private Long clusterId;

  //todo: currently topology is only associated with a single bp
  //todo: this will need to change to allow usage of multiple bp's for the same cluster
  //todo: for example: provision using bp1 and scale using bp2
  private Blueprint blueprint;
  private Configuration configuration;
  private ConfigRecommendationStrategy configRecommendationStrategy;
  private ProvisionAction provisionAction = ProvisionAction.INSTALL_AND_START;
  private Map<String, AdvisedConfiguration> advisedConfigurations = new HashMap<>();
  private final Map<String, HostGroupInfo> hostGroupInfoMap = new HashMap<>();
  private final AmbariContext ambariContext;
  private final String defaultPassword;

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


  //todo: will need to convert all usages of hostgroup name to use fully qualified name (BP/HG)
  //todo: for now, restrict scaling to the same BP
  public ClusterTopologyImpl(AmbariContext ambariContext, TopologyRequest topologyRequest) throws InvalidTopologyException {
    this.clusterId = topologyRequest.getClusterId();
    // provision cluster currently requires that all hostgroups have same BP so it is ok to use root level BP here
    this.blueprint = topologyRequest.getBlueprint();
    this.configuration = topologyRequest.getConfiguration();
    if (topologyRequest instanceof ProvisionClusterRequest) {
      this.defaultPassword = ((ProvisionClusterRequest) topologyRequest).getDefaultPassword();
    } else {
      this.defaultPassword = null;
    }

    registerHostGroupInfo(topologyRequest.getHostGroupInfo());

    this.ambariContext = ambariContext;
  }

  @Override
  public void update(TopologyRequest topologyRequest) throws InvalidTopologyException {
    registerHostGroupInfo(topologyRequest.getHostGroupInfo());
  }

  @Override
  public Long getClusterId() {
    return clusterId;
  }

  @Override
  public void setClusterId(Long clusterId) {
    this.clusterId = clusterId;
  }

  @Override
  public Blueprint getBlueprint() {
    return blueprint;
  }

  @Override
  public Configuration getConfiguration() {
    return configuration;
  }

  @Override
  public Map<String, HostGroupInfo> getHostGroupInfo() {
    return hostGroupInfoMap;
  }

  //todo: do we want to return groups with no requested hosts?
  @Override
  public Collection<String> getHostGroupsForComponent(String component) {
    Collection<String> resultGroups = new ArrayList<>();
    for (HostGroup group : getBlueprint().getHostGroups().values() ) {
      if (group.getComponentNames().contains(component)) {
        resultGroups.add(group.getName());
      }
    }
    return resultGroups;
  }

  @Override
  public String getHostGroupForHost(String hostname) {
    for (HostGroupInfo groupInfo : hostGroupInfoMap.values() ) {
      if (groupInfo.getHostNames().contains(hostname)) {
        // a host can only be associated with a single host group
        return groupInfo.getHostGroupName();
      }
    }
    return null;
  }

  //todo: host info?
  @Override
  public void addHostToTopology(String hostGroupName, String host) throws InvalidTopologyException, NoSuchHostGroupException {
    if (blueprint.getHostGroup(hostGroupName) == null) {
      throw new NoSuchHostGroupException("Attempted to add host to non-existing host group: " + hostGroupName);
    }

    // check for host duplicates
    String groupContainsHost = getHostGroupForHost(host);
    // in case of reserved host, hostgroup will already contain host
    if (groupContainsHost != null && ! hostGroupName.equals(groupContainsHost)) {
      throw new InvalidTopologyException(String.format(
          "Attempted to add host '%s' to hostgroup '%s' but it is already associated with hostgroup '%s'.",
          host, hostGroupName, groupContainsHost));
    }

    synchronized(hostGroupInfoMap) {
      HostGroupInfo existingHostGroupInfo = hostGroupInfoMap.get(hostGroupName);
      if (existingHostGroupInfo == null) {
        throw new RuntimeException(String.format("An attempt was made to add host '%s' to an unregistered hostgroup '%s'",
            host, hostGroupName));
      }
      // ok to add same host multiple times to same group
      existingHostGroupInfo.addHost(host);

      LOG.info("ClusterTopologyImpl.addHostTopology: added host = " + host + " to host group = " + existingHostGroupInfo.getHostGroupName());
    }
  }

  @Override
  public Set<String> getAllHosts() {
    return hostGroupInfoMap.values().stream().flatMap(hg -> hg.getHostNames().stream()).collect(toSet());
  }

  @Override
  public Collection<String> getHostAssignmentsForComponent(String component) {
    //todo: ordering requirements?
    Collection<String> hosts = new ArrayList<>();
    Collection<String> hostGroups = getHostGroupsForComponent(component);
    for (String group : hostGroups) {
      HostGroupInfo hostGroupInfo = getHostGroupInfo().get(group);
      if (hostGroupInfo != null) {
        hosts.addAll(hostGroupInfo.getHostNames());
      } else {
        LOG.warn("HostGroup {} not found, when checking for hosts for component {}", group, component);
      }
    }
    return hosts;
  }

  @Override
  public boolean isNameNodeHAEnabled() {
    return isNameNodeHAEnabled(configuration.getFullProperties());
  }

  public static boolean isNameNodeHAEnabled(Map<String, Map<String, String>> configurationProperties) {
    return configurationProperties.containsKey("hdfs-site") &&
           (configurationProperties.get("hdfs-site").containsKey("dfs.nameservices") ||
            configurationProperties.get("hdfs-site").containsKey("dfs.internal.nameservices"));
  }

  @Override
  public boolean isYarnResourceManagerHAEnabled() {
    return isYarnResourceManagerHAEnabled(configuration.getFullProperties());
  }

  /**
   * Static convenience function to determine if Yarn ResourceManager HA is enabled
   * @param configProperties configuration properties for this cluster
   * @return true if Yarn ResourceManager HA is enabled
   *         false if Yarn ResourceManager HA is not enabled
   */
  static boolean isYarnResourceManagerHAEnabled(Map<String, Map<String, String>> configProperties) {
    return configProperties.containsKey("yarn-site") && configProperties.get("yarn-site").containsKey("yarn.resourcemanager.ha.enabled")
      && configProperties.get("yarn-site").get("yarn.resourcemanager.ha.enabled").equals("true");
  }

  @Override
  public boolean isClusterKerberosEnabled() {
    return ambariContext.isClusterKerberosEnabled(getClusterId());
  }

  @Override
  public RequestStatusResponse installHost(String hostName, boolean skipInstallTaskCreate, boolean skipFailure) {
    try {
      String hostGroupName = getHostGroupForHost(hostName);
      HostGroup hostGroup = this.blueprint.getHostGroup(hostGroupName);

      Collection<String> skipInstallForComponents = new ArrayList<>();
      if (skipInstallTaskCreate) {
        skipInstallForComponents.add(ALL_COMPONENTS);
      } else {
        // get the set of components that are marked as START_ONLY for this hostgroup
        skipInstallForComponents.addAll(hostGroup.getComponentNames(ProvisionAction.START_ONLY));
      }

      Collection<String> dontSkipInstallForComponents = hostGroup.getComponentNames(INSTALL_ONLY);
      dontSkipInstallForComponents.addAll(hostGroup.getComponentNames(INSTALL_AND_START));

      return ambariContext.installHost(hostName, ambariContext.getClusterName(getClusterId()),
        skipInstallForComponents, dontSkipInstallForComponents, skipFailure);
    } catch (AmbariException e) {
      LOG.error("Cannot get cluster name for clusterId = " + getClusterId(), e);
      throw new RuntimeException(e);
    }
  }

  @Override
  public RequestStatusResponse startHost(String hostName, boolean skipFailure) {
    try {
      String hostGroupName = getHostGroupForHost(hostName);
      HostGroup hostGroup = this.blueprint.getHostGroup(hostGroupName);

      // get the set of components that are marked as INSTALL_ONLY
      // for this hostgroup
      Collection<String> installOnlyComponents =
        hostGroup.getComponentNames(ProvisionAction.INSTALL_ONLY);

      return ambariContext.startHost(hostName, ambariContext.getClusterName(getClusterId()), installOnlyComponents, skipFailure);
    } catch (AmbariException e) {
      LOG.error("Cannot get cluster name for clusterId = " + getClusterId(), e);
      throw new RuntimeException(e);
    }
  }

  @Override
  public void setConfigRecommendationStrategy(ConfigRecommendationStrategy strategy) {
    this.configRecommendationStrategy = strategy;
  }

  @Override
  public ConfigRecommendationStrategy getConfigRecommendationStrategy() {
    return this.configRecommendationStrategy;
  }

  @Override
  public ProvisionAction getProvisionAction() {
    return provisionAction;
  }

  @Override
  public void setProvisionAction(ProvisionAction provisionAction) {
    this.provisionAction = provisionAction;
  }

  @Override
  public Map<String, AdvisedConfiguration> getAdvisedConfigurations() {
    return this.advisedConfigurations;
  }

  @Override
  public AmbariContext getAmbariContext() {
    return ambariContext;
  }

  @Override
  public void removeHost(String hostname) {
    for(Map.Entry<String,HostGroupInfo> entry : hostGroupInfoMap.entrySet()) {
      entry.getValue().removeHost(hostname);
    }
  }

  @Override
  public String getDefaultPassword() {
    return defaultPassword;
  }

  @Override
  public boolean hasHadoopCompatibleService() {
    return blueprint.getServiceInfos().stream()
      .anyMatch(service -> HADOOP_COMPATIBLE_FS.equals(service.getServiceType()));
  }

  private void registerHostGroupInfo(Map<String, HostGroupInfo> requestedHostGroupInfoMap) throws InvalidTopologyException {
    LOG.debug("Registering requested host group information for {} hostgroups", requestedHostGroupInfoMap.size());
    checkForDuplicateHosts(requestedHostGroupInfoMap);

    for (HostGroupInfo requestedHostGroupInfo : requestedHostGroupInfoMap.values()) {
      String hostGroupName = requestedHostGroupInfo.getHostGroupName();

      //todo: doesn't support using a different blueprint for update (scaling)
      HostGroup baseHostGroup = getBlueprint().getHostGroup(hostGroupName);

      if (baseHostGroup == null) {
        throw new IllegalArgumentException("Invalid host_group specified: " + hostGroupName +
            ".  All request host groups must have a corresponding host group in the specified blueprint");
      }
      //todo: split into two methods
      HostGroupInfo currentHostGroupInfo = hostGroupInfoMap.get(hostGroupName);
      if (currentHostGroupInfo == null) {
        // blueprint host group config
        Configuration bpHostGroupConfig = baseHostGroup.getConfiguration();
        // parent config is BP host group config but with parent set to topology cluster scoped config
        Configuration parentConfiguration = new Configuration(bpHostGroupConfig.getProperties(),
            bpHostGroupConfig.getAttributes(), getConfiguration());

        requestedHostGroupInfo.getConfiguration().setParentConfiguration(parentConfiguration);
        hostGroupInfoMap.put(hostGroupName, requestedHostGroupInfo);
      } else {
        // Update.  Either add hosts or increment request count
        if (!requestedHostGroupInfo.getHostNames().isEmpty()) {
          try {
            // this validates that hosts aren't already registered with groups
            addHostsToTopology(requestedHostGroupInfo);
          } catch (NoSuchHostGroupException e) {
            //todo
            throw new InvalidTopologyException("Attempted to add hosts to unknown host group: " + hostGroupName);
          }
        } else {
          currentHostGroupInfo.setRequestedCount(
              currentHostGroupInfo.getRequestedHostCount() + requestedHostGroupInfo.getRequestedHostCount());
        }
        //todo: throw exception in case where request attempts to modify HG configuration in scaling operation
      }
    }
  }

  private void addHostsToTopology(HostGroupInfo hostGroupInfo) throws InvalidTopologyException, NoSuchHostGroupException {
    for (String host: hostGroupInfo.getHostNames()) {
      registerRackInfo(hostGroupInfo, host);
      addHostToTopology(hostGroupInfo.getHostGroupName(), host);
    }
  }

  private void registerRackInfo(HostGroupInfo hostGroupInfo, String host) {
    synchronized (hostGroupInfoMap) {
      HostGroupInfo cachedHGI = hostGroupInfoMap.get(hostGroupInfo.getHostGroupName());
      if (null != cachedHGI) {
        cachedHGI.addHostRackInfo(host, hostGroupInfo.getHostRackInfo().get(host));
      }
    }
  }


  private void checkForDuplicateHosts(Map<String, HostGroupInfo> groupInfoMap) throws InvalidTopologyException {
    Set<String> hosts = new HashSet<>();
    Set<String> duplicates = new HashSet<>();
    for (HostGroupInfo group : groupInfoMap.values()) {
      // check for duplicates within the new groups
      Collection<String> groupHosts = group.getHostNames();
      Collection<String> groupHostsCopy = new HashSet<>(group.getHostNames());
      groupHostsCopy.retainAll(hosts);
      duplicates.addAll(groupHostsCopy);
      hosts.addAll(groupHosts);

      // check against existing groups
      for (String host : groupHosts) {
        if (getHostGroupForHost(host) != null) {
          duplicates.add(host);
        }
      }
    }
    if (! duplicates.isEmpty()) {
      throw new InvalidTopologyException("The following hosts are mapped to multiple host groups: " + duplicates + "." +
        " Be aware that host names are converted to lowercase, case differences do not matter in Ambari deployments.");
    }
  }
}
