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


import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

import org.apache.ambari.annotations.UpgradeCheckInfo;
import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.actionmanager.HostRoleStatus;
import org.apache.ambari.server.controller.AmbariManagementController;
import org.apache.ambari.server.controller.AmbariServer;
import org.apache.ambari.server.controller.internal.AbstractControllerResourceProvider;
import org.apache.ambari.server.controller.internal.RequestResourceProvider;
import org.apache.ambari.server.controller.spi.Request;
import org.apache.ambari.server.controller.spi.Resource;
import org.apache.ambari.server.controller.spi.ResourceProvider;
import org.apache.ambari.server.controller.utilities.PropertyHelper;
import org.apache.ambari.server.orm.dao.HostRoleCommandDAO;
import org.apache.ambari.server.orm.dao.RequestDAO;
import org.apache.ambari.server.orm.entities.HostRoleCommandEntity;
import org.apache.ambari.server.orm.entities.RequestEntity;
import org.apache.ambari.spi.upgrade.UpgradeCheckDescription;
import org.apache.ambari.spi.upgrade.UpgradeCheckGroup;
import org.apache.ambari.spi.upgrade.UpgradeCheckRequest;
import org.apache.ambari.spi.upgrade.UpgradeCheckResult;
import org.apache.ambari.spi.upgrade.UpgradeCheckStatus;
import org.apache.ambari.spi.upgrade.UpgradeCheckType;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Sets;
import com.google.inject.Inject;
import com.google.inject.Singleton;

/**
 * This pre upgrade check verifies if the Ambari Metrics Hadoop Sink package version on all hosts is the expected one
 * corresponding to the stack version. For example, in HDP 3.0, the corresponding ambari-metrics-hadoop-sink version should
 * be 2.7.0.0.
 */
@Singleton
@UpgradeCheckInfo(
  group = UpgradeCheckGroup.REPOSITORY_VERSION)
public class AmbariMetricsHadoopSinkVersionCompatibilityCheck extends ClusterCheck  {

  @Inject
  private RequestDAO requestDAO;

  @Inject
  private HostRoleCommandDAO hostRoleCommandDAO;

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

  private enum PreUpgradeCheckStatus {SUCCESS, FAILED, RUNNING}

  static final String HADOOP_SINK_VERSION_NOT_SPECIFIED = "hadoop-sink-version-not-specified";

  static final String MIN_HADOOP_SINK_VERSION_PROPERTY_NAME = "min-hadoop-sink-version";
  static final String RETRY_INTERVAL_PROPERTY_NAME = "request-status-check-retry-interval";
  static final String NUM_TRIES_PROPERTY_NAME = "request-status-check-num-retries";

  /**
   * Total wait time for Ambari Server request time to finish => 2 mins.
   */
  private long retryInterval = 6000l; // 6 seconds sleep interval per retry.
  private int numTries = 20; // 20 times the check will try to see if request finished.

  static final UpgradeCheckDescription AMS_HADOOP_SINK_VERSION_COMPATIBILITY = new UpgradeCheckDescription("AMS_HADOOP_SINK_VERSION_COMPATIBILITY",
      UpgradeCheckType.HOST,
      "Ambari Metrics Hadoop Sinks need to be compatible with the stack version. This check ensures that compatibility.",
      new ImmutableMap.Builder<String, String>().put(UpgradeCheckDescription.DEFAULT,"Hadoop Sink version check failed. " +
        "To fix this, please upgrade 'ambari-metrics-hadoop-sink' package to %s on all the failed hosts")
        .put(AmbariMetricsHadoopSinkVersionCompatibilityCheck.HADOOP_SINK_VERSION_NOT_SPECIFIED, "Hadoop Sink version for pre-check not specified. " +
          "Please use 'min-hadoop-sink-version' property in upgrade pack to specify min hadoop sink version").build());

  /**
   * Constructor.
   */
  public AmbariMetricsHadoopSinkVersionCompatibilityCheck() {
    super(AMS_HADOOP_SINK_VERSION_COMPATIBILITY);
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public Set<String> getApplicableServices() {
    return Sets.newHashSet("AMBARI_METRICS", "HDFS");
  }

  @Override
  public UpgradeCheckResult perform(UpgradeCheckRequest prereqCheckRequest) throws AmbariException {
    UpgradeCheckResult result = new UpgradeCheckResult(this);
    String minHadoopSinkVersion = null;
    Map<String, String> checkProperties = prereqCheckRequest.getCheckConfigurations();

    if(checkProperties != null) {
      minHadoopSinkVersion = checkProperties.get(MIN_HADOOP_SINK_VERSION_PROPERTY_NAME);
      retryInterval = Long.parseLong(checkProperties.getOrDefault(RETRY_INTERVAL_PROPERTY_NAME, "6000"));
      numTries = Integer.parseInt(checkProperties.getOrDefault(NUM_TRIES_PROPERTY_NAME, "20"));
    }

    if (StringUtils.isEmpty(minHadoopSinkVersion)) {
      LOG.debug("Hadoop Sink version for pre-check not specified.");
      result.setStatus(UpgradeCheckStatus.FAIL);
      result.setFailReason(getFailReason(HADOOP_SINK_VERSION_NOT_SPECIFIED, result, prereqCheckRequest));
      return result;
    }

    LOG.debug("Properties : Hadoop Sink Version = {} , retryInterval = {}, numTries = {}", minHadoopSinkVersion, retryInterval, numTries);

    AmbariManagementController ambariManagementController = AmbariServer.getController();

    ResourceProvider provider = AbstractControllerResourceProvider.getResourceProvider(
      Resource.Type.Request,
      ambariManagementController
      );

    String clusterName = prereqCheckRequest.getClusterName();

    Set<String> hosts = ambariManagementController.getClusters()
      .getCluster(clusterName).getHosts("AMBARI_METRICS", "METRICS_MONITOR");

    if (CollectionUtils.isEmpty(hosts)) {
      LOG.warn("No hosts have the component METRICS_MONITOR.");
      result.setStatus(UpgradeCheckStatus.PASS);
      return result;
    }

    Set<Map<String, Object>> propertiesSet = new HashSet<>();
    Map<String, Object> properties = new LinkedHashMap<>();
    properties.put(RequestResourceProvider.REQUEST_CLUSTER_NAME_PROPERTY_ID, clusterName);

    Set<Map<String, Object>> filterSet = new HashSet<>();
    Map<String, Object> filterMap = new HashMap<>();
    filterMap.put(RequestResourceProvider.SERVICE_ID, "AMBARI_METRICS");
    filterMap.put(RequestResourceProvider.COMPONENT_ID, "METRICS_MONITOR");
    filterMap.put(RequestResourceProvider.HOSTS_ID, StringUtils.join(hosts,","));
    filterSet.add(filterMap);

    properties.put(RequestResourceProvider.REQUEST_RESOURCE_FILTER_ID, filterSet);
    propertiesSet.add(properties);

    Map<String, String> requestInfoProperties = new HashMap<>();
    requestInfoProperties.put(RequestResourceProvider.COMMAND_ID, "CHECK_HADOOP_SINK_VERSION");
    requestInfoProperties.put(RequestResourceProvider.CONTEXT, "Pre Upgrade check for Hadoop Metric Sink version");

    Request request = PropertyHelper.getCreateRequest(propertiesSet, requestInfoProperties);
    try {
      org.apache.ambari.server.controller.spi.RequestStatus response = provider.createResources(request);
      Resource responseResource = response.getRequestResource();
      String requestIdProp = PropertyHelper.getPropertyId("Requests", "id");
      long requestId = (long) responseResource.getPropertyValue(requestIdProp);
      LOG.debug("RequestId for AMS Hadoop Sink version compatibility pre check : " + requestId);

      Thread.sleep(retryInterval);
      PreUpgradeCheckStatus status;
      int retry = 0;
      LinkedHashSet<String> failedHosts = new LinkedHashSet<>();
      while ((status = pollRequestStatus(requestId, failedHosts)).equals(PreUpgradeCheckStatus.RUNNING)
        && retry++ < numTries) {
        if (retry != numTries) {
          Thread.sleep(retryInterval);
        }
      }

      if (status.equals(PreUpgradeCheckStatus.SUCCESS)) {
        result.setStatus(UpgradeCheckStatus.PASS);
      } else {
        result.setStatus(UpgradeCheckStatus.FAIL);
        result.setFailReason(String.format(getFailReason(result, prereqCheckRequest), minHadoopSinkVersion));
        result.setFailedOn(failedHosts);
      }
    } catch (Exception e) {
      LOG.error("Error running Pre Upgrade check for AMS Hadoop Sink compatibility. " + e);
      result.setStatus(UpgradeCheckStatus.FAIL);
    }

    return result;
  }

  /**
   * Get the status of the requestId and also the set of failed hosts if any.
   * @param requestId RequestId to track.
   * @param failedHosts populate this argument for failed hosts.
   * @return Status of the request.
   * @throws Exception
   */
  private PreUpgradeCheckStatus pollRequestStatus(long requestId, Set<String> failedHosts) throws Exception {

    List<RequestEntity> requestEntities = requestDAO.findByPks(Collections.singleton(requestId), true);
    if (requestEntities != null && requestEntities.size() > 0) {

      RequestEntity requestEntity = requestEntities.iterator().next();
      HostRoleStatus requestStatus = requestEntity.getStatus();

      if (HostRoleStatus.COMPLETED.equals(requestStatus)) {
        return PreUpgradeCheckStatus.SUCCESS;
      }

      else if (requestStatus.isFailedState()) {
        failedHosts.addAll(getPreUpgradeCheckFailedHosts(requestEntity));
        LOG.debug("Hadoop Sink version check failed on the following hosts : " + failedHosts.stream().collect(Collectors.joining(",")));
        return PreUpgradeCheckStatus.FAILED;
      } else {
        return PreUpgradeCheckStatus.RUNNING;
      }
    } else {
      LOG.error("Unable to find RequestEntity for created request.");
    }
    return PreUpgradeCheckStatus.FAILED;
  }

  /**
   * Get the set of hosts (tasks) that failed the check.
   * @param requestEntity request info to get the task level info.
   * @return Set of hosts which failed the check along with type of failed state.
   * @throws Exception
   */
  private Set<String> getPreUpgradeCheckFailedHosts(RequestEntity requestEntity) throws Exception {

    List<HostRoleCommandEntity> hostRoleCommandEntities = hostRoleCommandDAO.findByRequest(requestEntity.getRequestId(), true);

    Set<String> failedHosts = new LinkedHashSet<>();
    for (HostRoleCommandEntity hostRoleCommandEntity : hostRoleCommandEntities) {
      HostRoleStatus status = hostRoleCommandEntity.getStatus();
      if (status.isFailedState()) {
        failedHosts.add(hostRoleCommandEntity.getHostName() + "(" + status + ")");
      }
    }
    return failedHosts;
  }
}