/*
 * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
 * (the "License"). You may not use this work except in compliance with the License, which is
 * available at www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied, as more fully set forth in the License.
 *
 * See the NOTICE file distributed with this work for information regarding copyright ownership.
 */

package alluxio.cli.hdfs;

import alluxio.cli.AbstractValidationTask;
import alluxio.cli.ApplicableUfsType;
import alluxio.cli.ValidationTaskResult;
import alluxio.cli.ValidationUtils;
import alluxio.collections.Pair;
import alluxio.conf.AlluxioConfiguration;
import alluxio.conf.PropertyKey;
import alluxio.exception.InvalidPathException;
import alluxio.util.ExceptionUtils;
import alluxio.util.io.PathUtils;

import java.io.IOException;
import java.net.URI;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import javax.annotation.Nullable;

/**
 * Class for validating HDFS-related configurations.
 * Validates accessibility and correctness of the configuration files passed to Alluxio.
 */
@ApplicableUfsType(ApplicableUfsType.Type.HDFS)
public class HdfsConfValidationTask extends AbstractValidationTask {
  public static final String SEPARATOR = ":";
  protected final AlluxioConfiguration mConf;
  final String mPath;
  // loaded by loadHdfsConfig()
  Map<String, String> mCoreConf = null;
  Map<String, String> mHdfsConf = null;
  ValidationUtils.State mState = ValidationUtils.State.OK;
  StringBuilder mMsg = new StringBuilder();
  StringBuilder mAdvice = new StringBuilder();

  /**
   * Creates a new instance of {@link HdfsConfValidationTask}
   * for validating HDFS configuration.
   *
   * @param path the UFS path
   * @param conf the UFS configuration
   */
  public HdfsConfValidationTask(String path, AlluxioConfiguration conf) {
    mPath = path;
    mConf = conf;
  }

  @Override
  public String getName() {
    return "ValidateHdfsConf";
  }

  protected ValidationTaskResult loadHdfsConfig() {
    Pair<String, String> clientConfFiles = getHdfsConfPaths();
    String coreConfPath = clientConfFiles.getFirst();
    String hdfsConfPath = clientConfFiles.getSecond();

    mCoreConf = accessAndParseConf("core-site.xml", coreConfPath);
    mHdfsConf = accessAndParseConf("hdfs-site.xml", hdfsConfPath);
    return new ValidationTaskResult(mState, getName(), mMsg.toString(), mAdvice.toString());
  }

  protected Pair<String, String> getHdfsConfPaths() {
    // If Configuration does not contain the key, then a {@link RuntimeException} will be
    // thrown before calling the {@link String#split} method.
    String confVal = mConf.getString(PropertyKey.UNDERFS_HDFS_CONFIGURATION);
    String[] clientHadoopConfFilePaths = confVal.split(SEPARATOR);
    mMsg.append(String.format(
        "%d file path(s) detected in for HDFS configuration files for \"%s\"%n",
        clientHadoopConfFilePaths.length, confVal));

    String clientCoreSiteFilePath = null;
    String clientHdfsSiteFilePath = null;
    for (String path : clientHadoopConfFilePaths) {
      if (path.contains("core-site.xml")) {
        clientCoreSiteFilePath = path;
      } else if (path.contains("hdfs-site.xml")) {
        clientHdfsSiteFilePath = path;
      }
    }
    if (clientHadoopConfFilePaths.length < 2 || clientCoreSiteFilePath == null
        || clientHdfsSiteFilePath == null) {
      mAdvice.append(String.format("Additional HDFS configuration can be specified with your"
          + " Alluxio mount point by configuring the property %s. The value for this property "
          + "should be in the format \"{core-site.xml path}:{hdfs-site.xml path}\"%n",
          PropertyKey.UNDERFS_HDFS_CONFIGURATION.getName()));
    }
    return new Pair<>(clientCoreSiteFilePath, clientHdfsSiteFilePath);
  }

  @Override
  public ValidationTaskResult validateImpl(Map<String, String> optionsMap) {
    if (!ValidationUtils.isHdfsScheme(mPath)) {
      mMsg.append(String.format(
              "UFS path %s is not HDFS. Skipping validation for HDFS properties.%n", mPath));
      return new ValidationTaskResult(ValidationUtils.State.SKIPPED, getName(),
              mMsg.toString(), mAdvice.toString());
    }

    ValidationTaskResult loadConfig = loadHdfsConfig();
    if (loadConfig.getState() != ValidationUtils.State.OK) {
      // If failed to load config files, abort
      return loadConfig;
    }

    // no conflicts between these two
    ValidationTaskResult last = checkConflicts();
    if (last.getState() == ValidationUtils.State.OK) {
      last = checkNameservice();
    }

    return last;
  }

  protected ValidationTaskResult checkNameservice() {
    ValidationUtils.State state;
    String nsKey = "dfs.nameservices";
    String nameservices = mCoreConf.get(nsKey);
    if (nameservices == null) {
      nameservices = mHdfsConf.get(nsKey);
    }
    if (nameservices == null) {
      return new ValidationTaskResult(ValidationUtils.State.OK, getName(), "No nameservice "
          + "detected", "");
    }
    List<String> nsList =
        Arrays.stream(nameservices.split(",")).map(String::trim).collect(Collectors.toList());

    try {
      URI u = URI.create(mPath);
      String uriHost = u.getHost().toLowerCase();
      long nsCount = nsList.stream().filter(s -> uriHost.equals(s.toLowerCase())).count();
      state = ValidationUtils.State.OK;
      if (nsCount < 1) {
        state = ValidationUtils.State.FAILED;
        mAdvice.append(String.format("One or more nameservices (%s) were detected in the HDFS "
            + "configuration, but not used in the URI to connect to HDFS.", nameservices));
        mMsg.append(String.format("Could not find any of the configured nameservices (%s) in the "
            + "given HDFS connection URI (%s)", nameservices, u));
      }
    } catch (IllegalArgumentException e) {
      state = ValidationUtils.State.FAILED;
      mMsg.append("HDFS path not parsable as a URI.");
      mMsg.append(ExceptionUtils.asPlainText(e));
      mAdvice.append("Make sure the HDFS URI is in a valid format.");
    }
    return new ValidationTaskResult(state, getName(), mMsg.toString(), mAdvice.toString());
  }

  // Verify core-site.xml and hdfs.site.xml has no conflicts
  // mCoreConf and mHdfsConf are verified to be non-null as precondition
  protected ValidationTaskResult checkConflicts() {
    ValidationUtils.State state = ValidationUtils.State.OK;
    for (Map.Entry<String, String> entry : mCoreConf.entrySet()) {
      String k = entry.getKey();
      if (mHdfsConf.containsKey(k)) {
        String hdfsValue = mHdfsConf.get(k);
        String coreValue = mCoreConf.get(k);
        if (!hdfsValue.equals(coreValue)) {
          state = ValidationUtils.State.FAILED;
          mMsg.append(String.format("Property %s is %s in core-site.xml and %s in hdfs-site.xml",
                  k, coreValue, hdfsValue));
          mAdvice.append(String.format(
                  "Please fix the inconsistency for %s in core-site.xml and hdfs.xml.%n", k));
        }
      }
    }
    if (state == ValidationUtils.State.OK) {
      mMsg.append("core-site.xml and hdfs-site.xml are consistent.\n");
    }
    return new ValidationTaskResult(state, getName(), mMsg.toString(), mAdvice.toString());
  }

  @Nullable
  private Map<String, String> accessAndParseConf(String configName, String path) {
    if (path == null || path.isEmpty()) {
      mMsg.append(String.format("%s is not configured in Alluxio property %s%n", configName,
              PropertyKey.UNDERFS_HDFS_CONFIGURATION));
      mState = ValidationUtils.State.SKIPPED;
      return null;
    }
    try {
      PathUtils.getPathComponents(path);
    } catch (InvalidPathException e) {
      mState = ValidationUtils.State.WARNING;
      mMsg.append(String.format("Invalid path %s in Alluxio property %s.%n", path,
              PropertyKey.UNDERFS_HDFS_CONFIGURATION));
      mMsg.append(ExceptionUtils.asPlainText(e));
      mAdvice.append(String.format("Please correct the path for %s in %s%n", configName,
              PropertyKey.UNDERFS_HDFS_CONFIGURATION));
      return null;
    }
    Path confPath = Paths.get(path);
    if (!Files.exists(confPath)) {
      mState = ValidationUtils.State.WARNING;
      mMsg.append(String.format("File does not exist at %s in Alluxio property %s.%n", path,
          PropertyKey.UNDERFS_HDFS_CONFIGURATION));
      mAdvice.append(String.format("Could not file file at \"%s\". Correct the path in %s%n",
          configName, PropertyKey.UNDERFS_HDFS_CONFIGURATION));
      return null;
    }
    if (!Files.isReadable(Paths.get(path))) {
      mState = ValidationUtils.State.WARNING;
      mMsg.append(String.format("\"%s\" is not readable from Alluxio property %s.%n",
          path, PropertyKey.UNDERFS_HDFS_CONFIGURATION));
      mAdvice.append(String.format("Grant more accessible permissions on the file"
              + " %s from %s%n",
          configName, PropertyKey.UNDERFS_HDFS_CONFIGURATION));
      return null;
    }
    HadoopConfigurationFileParser parser = new HadoopConfigurationFileParser();
    Map<String, String> properties = null;
    try {
      properties = parser.parseXmlConfiguration(path);
      mMsg.append(String.format("Successfully loaded %s. %n", path));
    } catch (IOException e) {
      mState = ValidationUtils.State.FAILED;
      mMsg.append(String.format("Failed to read %s. %s.%n", path, e.getMessage()));
      mMsg.append(ExceptionUtils.asPlainText(e));
      mAdvice.append(String.format("Please check your %s.%n", path));
    } catch (RuntimeException e) {
      mState = ValidationUtils.State.FAILED;
      mMsg.append(String.format("Failed to parse %s. %s.%n", path, e.getMessage()));
      mMsg.append(ExceptionUtils.asPlainText(e));
      mAdvice.append(String.format("Failed to parse %s as valid XML. Please check that the file "
          + "path is correct and the content is valid XML.%n", path));
    }
    return properties;
  }
}
