alluxio-master-2.x/integration/tools/validation/src/main/java/alluxio/cli/UnderFileSystemContractTest.java
/*
 * 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;

import static alluxio.conf.PropertyKey.PropertyType.STRING;

import alluxio.conf.Configuration;
import alluxio.conf.InstancedConfiguration;
import alluxio.conf.PropertyKey;
import alluxio.conf.Source;
import alluxio.underfs.UfsFileStatus;
import alluxio.underfs.UfsStatus;
import alluxio.underfs.UnderFileSystem;
import alluxio.underfs.UnderFileSystemConfiguration;
import alluxio.underfs.UnderFileSystemFactory;
import alluxio.underfs.UnderFileSystemFactoryRegistry;
import alluxio.underfs.options.DeleteOptions;
import alluxio.util.ExceptionUtils;
import alluxio.util.io.PathUtils;

import com.beust.jcommander.JCommander;
import com.beust.jcommander.Parameter;
import com.google.common.io.Closer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.PrintStream;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.stream.Collectors;

/**
 * Integration tests for Alluxio under filesystems. It describes the contract of Alluxio
 * with the UFS through the UFS interface.
 *
 * This class will run all tests (with "Test" suffix in the method name)
 * in {@link UnderFileSystemCommonOperations}. If the given ufs path is a S3 path,
 * all tests in {@link S3ASpecificOperations} will also be run.
 */
public final class UnderFileSystemContractTest {
  private static final Logger LOG = LoggerFactory.getLogger(UnderFileSystemContractTest.class);

  public static final String TASK_NAME = "ValidateUfsOperations";
  private static final String S3_IDENTIFIER = "s3";

  @Parameter(names = {"--path"}, required = true,
      description = "The under filesystem path to run tests against.")
  private String mUfsPath;

  @Parameter(names = {"--test"}, required = false,
      description = "Test name, this option can be passed multiple times to "
          + "indicate multiply tests")
  private List<String> mTestList = new ArrayList<>();

  @Parameter(names = {"--help"}, help = true)
  private boolean mHelp = false;

  private InstancedConfiguration mConf;
  private UnderFileSystem mUfs;

  /**
   * A constructor from default.
   * */
  public UnderFileSystemContractTest() {
    mConf = Configuration.modifiableGlobal();
  }

  /**
   * Runs the tests and returns nothing.
   * */
  public void run() throws Exception {
    UnderFileSystemConfiguration ufsConf = getUfsConf();
    UnderFileSystemFactory factory = UnderFileSystemFactoryRegistry.find(mUfsPath, ufsConf);
    // Check if the ufs path is valid
    if (factory == null || !factory.supportsPath(mUfsPath)) {
      System.out.printf("%s is not a valid path", mUfsPath);
      System.exit(1);
    }

    // Set common properties
    mConf.set(PropertyKey.UNDERFS_LISTING_LENGTH, 50);
    mConf.set(PropertyKey.USER_BLOCK_SIZE_BYTES_DEFAULT, "512B");
    // Increase the buffer time of journal writes to speed up tests
    mConf.set(PropertyKey.MASTER_JOURNAL_FLUSH_BATCH_TIME_MS, "1sec");

    mUfs = UnderFileSystem.Factory.create(mUfsPath, ufsConf);

    int failedCnt = runCommonOperations();

    if (mUfs.getUnderFSType().equals(S3_IDENTIFIER)) {
      failedCnt += runS3Operations();
    }
    System.out.printf("Tests completed with %d failed.%n", failedCnt);
    if (failedCnt > 0) {
      throw new RuntimeException(String.format("UFS contract test failed with %d failures",
          failedCnt));
    }
  }

  /**
   * Runs the tests and return a {@link ValidationTaskResult}.
   *
   * @return a task result for all UFS tests
   * */
  // TODO(jiacheng): Refactor and move this into ValidateHdfsMount
  public ValidationTaskResult runValidationTask() throws IOException {
    Closer closer = Closer.create();
    final ByteArrayOutputStream msgBuf = new ByteArrayOutputStream();
    final ByteArrayOutputStream adviceBuf = new ByteArrayOutputStream();
    PrintStream msgStream = new PrintStream(msgBuf, true);
    PrintStream adviceStream = new PrintStream(adviceBuf, true);
    closer.register(msgStream);
    closer.register(adviceStream);
    closer.register(msgBuf);
    closer.register(adviceBuf);
    try {
      UnderFileSystemConfiguration ufsConf = getUfsConf();
      UnderFileSystemFactory factory = UnderFileSystemFactoryRegistry.find(mUfsPath, ufsConf);
      // Check if the ufs path is valid
      if (factory == null || !factory.supportsPath(mUfsPath)) {
        msgStream.append(String.format("%s is not a valid path%n", mUfsPath));
        adviceStream.append(String.format("Please validate if %s is a correct path%n", mUfsPath));
        return new ValidationTaskResult(ValidationUtils.State.FAILED, TASK_NAME,
                msgBuf.toString(), adviceBuf.toString());
      }

      // Set common properties
      mConf.set(PropertyKey.UNDERFS_LISTING_LENGTH, 50);
      mConf.set(PropertyKey.USER_BLOCK_SIZE_BYTES_DEFAULT, "512B");
      // Increase the buffer time of journal writes to speed up tests
      mConf.set(PropertyKey.MASTER_JOURNAL_FLUSH_BATCH_TIME_MS, "1sec");

      mUfs = UnderFileSystem.Factory.create(mUfsPath, ufsConf);

      int failedCnt = runCommonOperations(msgStream, adviceStream, System.err);

      if (mUfs.getUnderFSType().equals(S3_IDENTIFIER)) {
        failedCnt += runS3Operations(msgStream, adviceStream, System.err);
      }
      msgStream.append(String.format("Tests completed with %d failed.%n", failedCnt));
      ValidationUtils.State state = failedCnt == 0 ? ValidationUtils.State.OK
              : ValidationUtils.State.FAILED;
      if (failedCnt > 0) {
        adviceStream.append("Please check the failed UFS operations from the output.");
      }
      return new ValidationTaskResult(state, TASK_NAME, msgBuf.toString(),
              adviceBuf.toString());
    } catch (Exception e) {
      msgStream.append(ExceptionUtils.asPlainText(e));
      adviceStream.append("Please resolve the errors from failed UFS operations.");
      return new ValidationTaskResult(ValidationUtils.State.FAILED, TASK_NAME,
              msgBuf.toString(), adviceBuf.toString());
    } finally {
      closer.close();
    }
  }

  private UnderFileSystemConfiguration getUfsConf() {
    return UnderFileSystemConfiguration.defaults(mConf)
        .createMountSpecificConf(mConf.getProperties().entrySet().stream()
            .filter(entry -> mConf.getSource(entry.getKey()) == Source.SYSTEM_PROPERTY)
            .filter(entry -> mConf.isSet(entry.getKey()) && (entry.getKey().getType() != STRING
                || !((String) entry.getValue()).isEmpty()))
            .collect(Collectors.toMap(entry -> entry.getKey().getName(), Map.Entry::getValue)));
  }

  private int runCommonOperations() throws Exception {
    return runCommonOperations(System.out, System.out, System.err);
  }

  private int runCommonOperations(PrintStream msgStream, PrintStream adviceStream,
                                  PrintStream errStream) throws Exception {
    String testDir = createTestDirectory();
    return loadAndRunTests(new UnderFileSystemCommonOperations(mUfsPath, testDir, mUfs, mConf),
            testDir, msgStream, adviceStream, errStream);
  }

  private int runS3Operations() throws Exception {
    return runS3Operations(System.out, System.out, System.err);
  }

  private int runS3Operations(PrintStream msgStream,
                              PrintStream adviceStream, PrintStream errStream) throws Exception {
    mConf.set(PropertyKey.UNDERFS_S3_LIST_OBJECTS_V1, true);
    mConf.set(PropertyKey.UNDERFS_S3_STREAMING_UPLOAD_ENABLED, true);
    mConf.set(PropertyKey.UNDERFS_S3_STREAMING_UPLOAD_PARTITION_SIZE, "5MB");
    mConf.set(PropertyKey.UNDERFS_S3_INTERMEDIATE_UPLOAD_CLEAN_AGE, "0");

    mUfs = UnderFileSystem.Factory.create(mUfsPath, getUfsConf());

    String testDir = createTestDirectory();
    return loadAndRunTests(new S3ASpecificOperations(testDir, mUfs, mConf),
            testDir, msgStream, adviceStream, errStream);
  }

  private int loadAndRunTests(Object operations, String testDir) throws Exception {
    return loadAndRunTests(operations, testDir, System.out, System.out, System.err);
  }

    /**
     * Loads and runs the tests in the target operations class.
     *
     * @param operations the class that contains the tests to run
     * @param testDir the test directory to run tests against
     * @return the number of failed tests
     */
  private int loadAndRunTests(Object operations, String testDir, PrintStream msgStream,
                              PrintStream adviceStream, PrintStream errStream) throws Exception {
    int failedTestCnt = 0;
    try {
      Class classToRun = operations.getClass();
      Field[] fields = classToRun.getDeclaredFields();
      for (Field field : fields) {
        field.setAccessible(true);
      }
      Method[] tests = classToRun.getDeclaredMethods();
      for (Method test : tests) {
        String testName = test.getName();
        if (testName.endsWith("Test")) {
          if (!(mTestList.isEmpty() || mTestList.contains(testName))) {
            continue;
          }
          msgStream.format("Running test: %s...%n", testName);
          boolean passed = false;
          try {
            test.invoke(operations);
            passed = true;
          } catch (Exception e) {
            if (mUfs.getUnderFSType().equals(S3_IDENTIFIER)) {
              logRelatedS3Operations(test, msgStream);
            }
            msgStream.format("Operation %s failed%n", testName);
            msgStream.format(ExceptionUtils.asPlainText(e));
            errStream.format("Test %s.%s aborted%n%s%n", test.getClass(), test.getName(), e);
          } finally {
            cleanupUfs(testDir);
            RunTestUtils.printTestStatus(passed, msgStream);
            if (!passed) {
              failedTestCnt++;
            }
          }
        }
      }
      return failedTestCnt;
    } finally {
      mUfs.deleteDirectory(testDir, DeleteOptions.defaults().setRecursive(true));
      mUfs.close();
    }
  }

  /**
   * @return the test directory to run tests against
   */
  private String createTestDirectory() throws IOException {
    String testDir = PathUtils.concatPath(mUfsPath, UUID.randomUUID());
    mUfs.mkdirs(testDir);
    return testDir;
  }

  /**
   * Cleans all the files or sub directories inside the given directory
   * in the under filesystem.
   *
   * @param directory the directory to clean
   */
  private void cleanupUfs(String directory) throws IOException {
    UfsStatus[] statuses = mUfs.listStatus(directory);
    if (statuses == null) {
      LOG.error("Path {} is invalid.", directory);
      return;
    }
    for (UfsStatus status : statuses) {
      if (status instanceof UfsFileStatus) {
        mUfs.deleteFile(PathUtils.concatPath(directory, status.getName()));
      } else {
        mUfs.deleteDirectory(PathUtils.concatPath(directory, status.getName()),
            DeleteOptions.defaults().setRecursive(true));
      }
    }
  }

  /**
   * Logs the S3 operations related to the failed test. This method
   * should only be called when the ufs is S3A.
   *
   * @param test the test to log
   */
  private void logRelatedS3Operations(Method test, PrintStream stream) {
    RelatedS3Operations annotation = test.getAnnotation(RelatedS3Operations.class);
    if (annotation != null) {
      String[] ops = annotation.operations();
      if (ops.length > 0) {
        stream.println("Related S3 operations: " + String.join(", ", ops));
      }
    }
  }

  private static String getHelpMessage() {
    return "Test description:\n"
        + "Test the integration between Alluxio and the under filesystem. "
        + "If the given under filesystem is S3, this test can also be used as "
        + "a S3 compatibility test to test if the target under filesystem can "
        + "fulfill the minimum S3 compatibility requirements in order to "
        + "work well with Alluxio through Alluxio's integration with S3. \n"
        + "Command line example: 'bin/alluxio runUfsTests --path s3://testPath "
        + "-Ds3a.accessKeyId=<accessKeyId> -Ds3a.secretKeyId=<secretKeyId>"
        + "-Dalluxio.underfs.s3.endpoint=<endpoint_url> "
        + "-Dalluxio.underfs.s3.disable.dns.buckets=true'";
  }

  /**
   * @param args the input arguments
   */
  public static void main(String[] args) throws Exception {
    UnderFileSystemContractTest test = new UnderFileSystemContractTest();
    JCommander jc = new JCommander(test);
    jc.setProgramName(UnderFileSystemContractTest.class.getName());
    try {
      jc.parse(args);
    } catch (Exception e) {
      System.out.println(e.getMessage());
      jc.usage();
      System.out.println(getHelpMessage());
      System.exit(1);
    }
    if (test.mHelp) {
      jc.usage();
      System.out.println(getHelpMessage());
    } else {
      test.run();
    }
  }
}
