content: /*
   * 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.ApplicableUfsType;
  import alluxio.cli.ValidationTaskResult;
  import alluxio.cli.ValidationUtils;
  import alluxio.conf.AlluxioConfiguration;
  import alluxio.util.ExceptionUtils;
  import alluxio.util.io.PathUtils;
  
  import org.apache.commons.cli.Option;
  
  import java.util.ArrayList;
  import java.util.List;
  import java.util.Map;
  
  /**
   * Compares HDFS configuration in Alluxio and in HDFS environment variables.
   * */
  @ApplicableUfsType(ApplicableUfsType.Type.HDFS)
  public class HdfsConfParityValidationTask extends HdfsConfValidationTask {
    /** Name of the environment variable to store the path to Hadoop config directory. */
    protected static final String HADOOP_CONF_DIR_ENV_VAR = "HADOOP_CONF_DIR";
  
    protected static final Option HADOOP_CONF_DIR_OPTION =
            Option.builder("hadoopConfDir").required(false).hasArg(true)
                    .desc("path to server-side hadoop conf dir").build();
  
    /**
     * Constructor.
     *
     * @param path the UFS path
     * @param conf the UFS configuration
     * */
    public HdfsConfParityValidationTask(String path, AlluxioConfiguration conf) {
      super(path, conf);
    }
  
    @Override
    public String getName() {
      return "ValidateHdfsServerAndClientConf";
    }
  
    @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());
      }
  
      return validateHdfsSettingParity(optionsMap);
    }
  
    @Override
    public List<Option> getOptionList() {
      List<Option> opts = new ArrayList<>();
      opts.add(HADOOP_CONF_DIR_OPTION);
      return opts;
    }
  
    private ValidationTaskResult validateHdfsSettingParity(Map<String, String> optionsMap) {
      String serverHadoopConfDirPath;
      if (optionsMap.containsKey(HADOOP_CONF_DIR_OPTION.getOpt())) {
        serverHadoopConfDirPath = optionsMap.get(HADOOP_CONF_DIR_OPTION.getOpt());
      } else {
        serverHadoopConfDirPath = System.getenv(HADOOP_CONF_DIR_ENV_VAR);
      }
      if (serverHadoopConfDirPath == null) {
        mMsg.append("Path to server-side hadoop configuration unspecified,"
                + " skipping validation for HDFS properties.");
        return new ValidationTaskResult(ValidationUtils.State.SKIPPED, getName(),
                mMsg.toString(), mAdvice.toString());
      }
      String serverCoreSiteFilePath = PathUtils.concatPath(serverHadoopConfDirPath,
              "/core-site.xml");
      String serverHdfsSiteFilePath = PathUtils.concatPath(serverHadoopConfDirPath,
              "/hdfs-site.xml");
  
      // Load client core-site and hdfs-site config
      ValidationTaskResult loadConfig = loadHdfsConfig();
      if (loadConfig.getState() != ValidationUtils.State.OK) {
        // If failed to load config files, abort
        return loadConfig;
      }
  
      boolean ok = compareConfigurations(serverCoreSiteFilePath,
               "core-site.xml", mCoreConf)
              && compareConfigurations(serverHdfsSiteFilePath,
               "hdfs-site.xml", mHdfsConf);
      return new ValidationTaskResult(ok ? ValidationUtils.State.OK
               : ValidationUtils.State.FAILED, getName(), mMsg.toString(), mAdvice.toString());
    }
  
    private boolean compareConfigurations(String serverConfigFilePath, String clientSiteName,
                                          Map<String, String> clientSiteProps) {
      HadoopConfigurationFileParser parser = new HadoopConfigurationFileParser();
  
      Map<String, String> serverSiteProps;
      try {
        serverSiteProps = parser.parseXmlConfiguration(serverConfigFilePath);
      } catch (Exception e) {
        mMsg.append(String.format("Failed to parse server-side %s.%n", serverConfigFilePath));
        mMsg.append(ExceptionUtils.asPlainText(e));
        mAdvice.append(String.format("Please fix the parsing error in %s.%n", serverConfigFilePath));
        return false;
      }
  
      boolean matches = true;
      for (Map.Entry<String, String> prop : clientSiteProps.entrySet()) {
        if (!serverSiteProps.containsKey(prop.getKey())) {
          matches = false;
          mMsg.append(String.format("%s is configured in %s, but not configured in %s.%n",
                  prop.getKey(), clientSiteName, serverConfigFilePath));
          mAdvice.append(String.format("Please configure property %s in %s.%n",
                  prop.getKey(), serverConfigFilePath));
        } else if (!prop.getValue().equals(serverSiteProps.get(prop.getKey()))) {
          matches = false;
          mMsg.append(String.format("%s is set to %s in %s, but to %s in %s.%n",
                  prop.getKey(), prop.getValue(), clientSiteName,
                  serverSiteProps.get(prop.getKey()), serverConfigFilePath));
          mAdvice.append(String.format("Please fix the inconsistency on property %s.%n",
                  prop.getKey()));
        }
      }
      if (!matches) {
        return false;
      }
      for (Map.Entry<String, String> prop : serverSiteProps.entrySet()) {
        if (!clientSiteProps.containsKey(prop.getKey())) {
          matches = false;
          mMsg.append(String.format("%s is configured in %s, but not configured in %s.%n",
                  prop.getKey(), serverConfigFilePath, clientSiteName));
          mAdvice.append(String.format("Please configure %s in %s.%n",
                  prop.getKey(), clientSiteName));
        } else if (!prop.getValue().equals(clientSiteProps.get(prop.getKey()))) {
          matches = false;
          mMsg.append(String.format("%s is set to %s in %s, but to %s in %s.%n",
                  prop.getKey(), prop.getValue(), prop.getValue(),
                  clientSiteProps.get(prop.getKey()), clientSiteName));
          mAdvice.append(String.format("Please fix the inconsistency on property %s.%n",
                  prop.getKey()));
        }
      }
      return matches;
    }
  }
  