alluxio-master-2.x/tests/src/test/java/alluxio/client/cli/job/DistributedCommandsStatsTest.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.client.cli.job;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;

import alluxio.AlluxioURI;
import alluxio.Constants;
import alluxio.UnderFileSystemFactoryRegistryRule;
import alluxio.client.WriteType;
import alluxio.client.file.FileSystem;
import alluxio.client.file.FileSystemTestUtils;
import alluxio.grpc.WritePType;
import alluxio.job.plan.load.LoadConfig;
import alluxio.job.plan.migrate.MigrateConfig;
import alluxio.job.plan.persist.PersistConfig;
import alluxio.job.util.JobTestUtils;
import alluxio.job.wire.Status;
import alluxio.metrics.MetricKey;
import alluxio.metrics.MetricsSystem;
import alluxio.testutils.underfs.sleeping.SleepingUnderFileSystemFactory;
import alluxio.testutils.underfs.sleeping.SleepingUnderFileSystemOptions;

import com.google.common.collect.Sets;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;

import java.util.Collections;
import java.util.HashSet;

/**
 * Tests stat counter values and output of CANCEL operations for distributed commands.
 * The tests will depend on the timing of the actual job runs.
 * If the job completes fast enough before the CANCEL operations runs,then the test would fail.
 * The tests compare the job statuses (CANCEL or not) and stat counter values for each status.
 */
public class DistributedCommandsStatsTest extends JobShellTest {
  private static final long SLEEP_MS = Constants.SECOND_MS * 15;
  private static final int TEST_TIMEOUT = 45;
  // When the task is finished, end the waiting. We can check the specific task status later.
  private static final HashSet<Status> FINISH_STATUS = Sets.newHashSet(Status.CANCELED,
      Status.FAILED, Status.COMPLETED);

  @ClassRule
  public static UnderFileSystemFactoryRegistryRule sUnderfilesystemfactoryregistry =
        new UnderFileSystemFactoryRegistryRule(new SleepingUnderFileSystemFactory(
            new SleepingUnderFileSystemOptions()
                .setIsDirectoryMs(SLEEP_MS)
                .setIsFileMs(SLEEP_MS)
                .setGetStatusMs(SLEEP_MS)
                .setListStatusMs(SLEEP_MS)
                .setListStatusWithOptionsMs(SLEEP_MS)
                .setExistsMs(SLEEP_MS)
                .setMkdirsMs(SLEEP_MS)));

  @Rule
  public TemporaryFolder mTempFolder = new TemporaryFolder();

  private String mLocalUfsPath;
  private FileSystem mFileSystem;

  @Before
  public void before() throws Exception {
    mLocalUfsPath = mTempFolder.getRoot().getAbsolutePath();
    sFileSystem.mount(new AlluxioURI("/mnt/"), new AlluxioURI("sleep://" + mLocalUfsPath));
  }

  @Test
  public void testCompleteStats() throws Exception {
    final int length = 10;
    FileSystemTestUtils.createByteFile(sFileSystem, "/test", WritePType.THROUGH, length);

    long jobId = sJobMaster.run(new LoadConfig("/test", 1, Collections.EMPTY_SET,
            Collections.EMPTY_SET, Collections.EMPTY_SET, Collections.EMPTY_SET, false));

    JobTestUtils
            .waitForJobStatus(sJobMaster, jobId, Sets.newHashSet(Status.COMPLETED), TEST_TIMEOUT);

    sJobShell.run("stat", "-v", Long.toString(jobId));

    String[] output = mOutput.toString().split("\n");
    assertEquals(String.format("ID: %s", jobId), output[0]);
    assertEquals(String.format("Name: Load"), output[1]);
    assertTrue(output[2].contains("Description: LoadConfig"));
    assertTrue(output[2].contains("/test"));
    assertEquals("Status: COMPLETED", output[3]);
    assertEquals("Task 0", output[4]);
    assertTrue(output[5].contains("\tWorker: "));
    assertEquals("\tStatus: COMPLETED", output[7]);

    double completedCount = MetricsSystem.getMetricValue(
            MetricKey.MASTER_JOB_DISTRIBUTED_LOAD_SUCCESS.getName()).getValue();
    double fileCount = MetricsSystem.getMetricValue(
            MetricKey.MASTER_JOB_DISTRIBUTED_LOAD_FILE_COUNT.getName()).getValue();
    double fileSize = MetricsSystem.getMetricValue(
            MetricKey.MASTER_JOB_DISTRIBUTED_LOAD_FILE_SIZE.getName()).getValue();

    //Metrics for Migrate job type
    double completedMigrateCount = MetricsSystem.getMetricValue(
            MetricKey.MASTER_MIGRATE_JOB_SUCCESS.getName()).getValue();
    double completedMigrateFileCount = MetricsSystem.getMetricValue(
            MetricKey.MASTER_MIGRATE_JOB_FILE_COUNT.getName()).getValue();
    double completedMigrateFileSize = MetricsSystem.getMetricValue(
            MetricKey.MASTER_MIGRATE_JOB_FILE_SIZE.getName()).getValue();

    //Metrics for Persist job type
    double completedPersistCount = MetricsSystem.getMetricValue(
            MetricKey.MASTER_ASYNC_PERSIST_SUCCESS.getName()).getValue();
    double completedPersistFileCount = MetricsSystem.getMetricValue(
            MetricKey.MASTER_ASYNC_PERSIST_FILE_COUNT.getName()).getValue();
    double completedPersistFileSize = MetricsSystem.getMetricValue(
            MetricKey.MASTER_ASYNC_PERSIST_FILE_SIZE.getName()).getValue();

    //test counters for distributed load on Complete status.
    assertEquals(completedCount, 1, 0); //distributedLoad operation count equals 1.
    assertEquals(fileCount, 1, 0); // file count equals 1.
    assertEquals(fileSize, length, 0); // file size equals $length.

    //test for other job types. Migrate counters, should all be 0.
    assertEquals(completedMigrateCount, 0, 0);
    assertEquals(completedMigrateFileCount, 0, 0);
    assertEquals(completedMigrateFileSize, 0, 0);

    //test AsyncPersist counters, should all be 0.
    assertEquals(completedPersistCount, 0, 0);
    assertEquals(completedPersistFileCount, 0, 0);
    assertEquals(completedPersistFileSize, 0, 0);
  }

  @Test
  public void testDistributedLoadCancelStats() throws Exception {
    FileSystemTestUtils.createByteFile(sFileSystem,  "/mnt/testFileNew",
            WritePType.THROUGH, 10);

    long jobId = sJobMaster.run(new LoadConfig("/mnt/testFileNew",
            1, Collections.EMPTY_SET,
            Collections.EMPTY_SET, Collections.EMPTY_SET, Collections.EMPTY_SET, false));

    sJobShell.run("cancel", Long.toString(jobId));

    JobTestUtils
        .waitForJobStatus(sJobMaster, jobId, FINISH_STATUS, TEST_TIMEOUT);

    sJobShell.run("stat", "-v", Long.toString(jobId));

    String[] output = mOutput.toString().split("\n");
    assertEquals(String.format("ID: %s", jobId), output[0]);
    assertEquals(String.format("Name: Load"), output[1]);
    assertTrue(output[2].contains("Description: LoadConfig"));
    assertTrue(output[2].contains("/mnt/testFileNew"));
    assertEquals("Status: CANCELED", output[3]);
    assertEquals("Task 0", output[4]);
    assertTrue(output[5].contains("\tWorker: "));
    assertEquals("\tStatus: CANCELED", output[7]);

    double cancelledCount = MetricsSystem.getMetricValue(
            MetricKey.MASTER_JOB_DISTRIBUTED_LOAD_CANCEL.getName()).getValue();

    assertEquals(cancelledCount, 1, 0);
  }

  @Test
  public void testDistributedCpCancelStats() throws Exception {
    FileSystemTestUtils.createByteFile(sFileSystem,  "/mnt/testFileSource",
            WritePType.THROUGH, 10);

    long jobId = sJobMaster.run(new MigrateConfig(
            "/mnt/testFileSource", "/mnt/testFileDest",
            WriteType.THROUGH, false));

    sJobShell.run("cancel", Long.toString(jobId));

    JobTestUtils
            .waitForJobStatus(sJobMaster, jobId, FINISH_STATUS, TEST_TIMEOUT);

    sJobShell.run("stat", "-v", Long.toString(jobId));

    String[] output = mOutput.toString().split("\n");
    assertEquals(String.format("ID: %s", jobId), output[0]);
    assertEquals(String.format("Name: Migrate"), output[1]);
    assertTrue(output[2].contains("Description: MigrateConfig"));
    assertTrue(output[2].contains("/mnt/testFileSource"));
    assertTrue(output[2].contains("/mnt/testFileDest"));
    assertEquals("Status: CANCELED", output[3]);
    assertEquals("Task 0", output[4]);
    assertTrue(output[5].contains("\tWorker: "));
    assertEquals("\tStatus: CANCELED", output[7]);

    double cancelledCount = MetricsSystem.getMetricValue(
            MetricKey.MASTER_MIGRATE_JOB_CANCEL.getName()).getValue();
    double failedCount = MetricsSystem.getMetricValue(
            MetricKey.MASTER_MIGRATE_JOB_FAIL.getName()).getValue();
    double completedCount = MetricsSystem.getMetricValue(
            MetricKey.MASTER_MIGRATE_JOB_SUCCESS.getName()).getValue();

    assertEquals(cancelledCount, 1, 0);
    assertEquals(failedCount, 0, 0);
    assertEquals(completedCount, 0, 0);
  }

  @Test
  public void testAsyncPersistCancelStats() throws Exception {
    FileSystemTestUtils.createByteFile(sFileSystem,  "/mnt/testFile",
            WritePType.MUST_CACHE, 10000);

    long jobId = sJobMaster.run(new PersistConfig("/mnt/testFile",
            0, false, "/mnt/testUfsPath"));

    sJobShell.run("cancel", Long.toString(jobId));

    JobTestUtils
            .waitForJobStatus(sJobMaster, jobId, FINISH_STATUS, TEST_TIMEOUT);

    sJobShell.run("stat", "-v", Long.toString(jobId));

    String[] output = mOutput.toString().split("\n");
    assertEquals(String.format("ID: %s", jobId), output[0]);
    assertEquals(String.format("Name: Persist"), output[1]);
    assertTrue(output[2].contains("Description: PersistConfig"));
    assertTrue(output[2].contains("/mnt/testFile"));
    assertEquals("Status: CANCELED", output[3]);
    assertEquals("Task 0", output[4]);
    assertTrue(output[5].contains("\tWorker: "));
    assertEquals("\tStatus: CANCELED", output[6]);

    double cancelledCount = MetricsSystem.getMetricValue(
            MetricKey.MASTER_ASYNC_PERSIST_CANCEL.getName()).getValue();
    double failedCount = MetricsSystem.getMetricValue(
            MetricKey.MASTER_ASYNC_PERSIST_FAIL.getName()).getValue();
    double completedCount = MetricsSystem.getMetricValue(
            MetricKey.MASTER_ASYNC_PERSIST_SUCCESS.getName()).getValue();

    assertEquals(cancelledCount, 1, 0);
    assertEquals(failedCount, 0, 0);
    assertEquals(completedCount, 0, 0);
  }
}
