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

import static java.util.stream.Collectors.toList;
import static org.hamcrest.Matchers.containsString;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;

import alluxio.AlluxioTestDirectory;
import alluxio.AlluxioURI;
import alluxio.ClientContext;
import alluxio.Constants;
import alluxio.SystemOutRule;
import alluxio.annotation.dora.DoraTestTodoItem;
import alluxio.client.WriteType;
import alluxio.client.file.FileOutStream;
import alluxio.client.file.FileSystem;
import alluxio.client.meta.RetryHandlingMetaMasterClient;
import alluxio.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.grpc.FileSystemMasterCommonPOptions;
import alluxio.grpc.SetAttributePOptions;
import alluxio.master.MasterClientContext;
import alluxio.master.SingleMasterInquireClient;
import alluxio.master.journal.JournalType;
import alluxio.master.journal.raft.RaftJournalSystem;
import alluxio.master.journal.raft.RaftJournalUtils;
import alluxio.master.journal.tool.JournalTool;
import alluxio.testutils.BaseIntegrationTest;
import alluxio.testutils.IntegrationTestUtils;
import alluxio.testutils.LocalAlluxioClusterResource;
import alluxio.util.io.PathUtils;

import org.apache.ratis.server.RaftServerConfigKeys;
import org.apache.ratis.server.storage.RaftStorage;
import org.apache.ratis.server.storage.StorageImplUtils;
import org.apache.ratis.statemachine.impl.SimpleStateMachineStorage;
import org.apache.ratis.statemachine.impl.SingleFileSnapshotInfo;
import org.hamcrest.Matchers;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;

import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Arrays;
import java.util.List;

/**
 * Tests for {@link JournalTool}.
 */
@DoraTestTodoItem(action = DoraTestTodoItem.Action.REMOVE, owner = "Jiacheng",
    comment = "journal no longer exists in dora")
@Ignore
public class JournalToolTest extends BaseIntegrationTest {
  private static final int CHECKPOINT_SIZE = 100;

  private final ByteArrayOutputStream mOutput = new ByteArrayOutputStream();

  @Rule
  public SystemOutRule mSystemOutRule = new SystemOutRule(mOutput);

  @Rule
  public LocalAlluxioClusterResource mLocalAlluxioClusterResource =
      new LocalAlluxioClusterResource.Builder()
          .setProperty(PropertyKey.MASTER_JOURNAL_TYPE, JournalType.UFS)
          .setProperty(PropertyKey.MASTER_JOURNAL_CHECKPOINT_PERIOD_ENTRIES, CHECKPOINT_SIZE)
          .setProperty(PropertyKey.MASTER_JOURNAL_LOG_SIZE_BYTES_MAX, "100")
          .setProperty(PropertyKey.USER_FILE_WRITE_TYPE_DEFAULT, WriteType.CACHE_THROUGH).build();

  private File mDumpDir;
  private FileSystem mFs;

  @Before
  public void before() throws IOException {
    mDumpDir = AlluxioTestDirectory.createTemporaryDirectory("journal_dump");
    mFs = mLocalAlluxioClusterResource.get().getClient();
  }

  @Test
  public void dumpSimpleUfsJournal() throws Throwable {
    // Create a test directory to trigger journaling.
    mFs.createDirectory(new AlluxioURI("/test"));
    // Run journal tool.
    JournalTool.main(new String[] {"-outputDir", mDumpDir.getAbsolutePath()});
    // Verify that a non-zero dump file exists.
    assertThat(mOutput.toString(), containsString(mDumpDir.getAbsolutePath()));
    assertNonemptyFileExists(PathUtils.concatPath(mDumpDir, "edits.txt"));
  }

  @Test
  @LocalAlluxioClusterResource.Config(
      confParams = {PropertyKey.Name.MASTER_JOURNAL_TYPE, "EMBEDDED"})
  public void dumpSimpleEmbeddedJournal() throws Throwable {
    // Create a test directory to trigger journaling.
    mFs.createDirectory(new AlluxioURI("/test"));
    mLocalAlluxioClusterResource.get().stopMasters();
    // Run journal tool.
    String masterJournalPath =
        mLocalAlluxioClusterResource.get().getLocalAlluxioMaster().getJournalFolder();
    JournalTool.main(
        new String[] {"-inputDir", masterJournalPath, "-outputDir", mDumpDir.getAbsolutePath()});
    // Verify that a non-zero dump file exists.
    assertThat(mOutput.toString(), containsString(mDumpDir.getAbsolutePath()));
    assertNonemptyFileExists(PathUtils.concatPath(mDumpDir, "edits.txt"));
  }

  @Test
  public void dumpHeapCheckpointFromUfsJournal() throws Throwable {
    for (String name : Arrays.asList("/pin", "/max_replication", "/async_persist", "/ttl")) {
      mFs.createFile(new AlluxioURI(name)).close();
    }
    mFs.setAttribute(new AlluxioURI("/pin"),
        SetAttributePOptions.newBuilder().setPinned(true).build());
    mFs.setAttribute(new AlluxioURI("/max_replication"),
        SetAttributePOptions.newBuilder().setReplicationMax(5).build());
    mFs.persist(new AlluxioURI("/async_persist"));
    mFs.setAttribute(new AlluxioURI("/ttl"),
        SetAttributePOptions.newBuilder()
            .setCommonOptions(FileSystemMasterCommonPOptions.newBuilder().setTtl(100000).build())
            .build());
    checkpointUfsJournal();
    JournalTool.main(new String[] {"-outputDir", mDumpDir.getAbsolutePath()});
    String checkpointDir = findCheckpointDir();

    assertNonemptyFileExists(PathUtils.concatPath(mDumpDir, "edits.txt"));
    assertNonemptyFileExists(PathUtils.concatPath(checkpointDir, "INODE_DIRECTORY_ID_GENERATOR"));
    for (String subPath : Arrays.asList("INODE_COUNTER", "PINNED_INODE_FILE_IDS",
        "REPLICATION_LIMITED_FILE_IDS", "TO_BE_PERSISTED_FILE_IDS")) {
      assertNonemptyFileExists(PathUtils.concatPath(checkpointDir, "INODE_TREE", subPath));
    }
  }

  @Test
  public void dumpBlockMasterCheckpointFromUfsJournal() throws Throwable {
    blockMasterCheckpointUfsJournal();
    JournalTool.main(new String[] {"-outputDir", mDumpDir.getAbsolutePath(),
        "-master", Constants.BLOCK_MASTER_NAME});

    assertNonemptyFileWithPrefixExist(mDumpDir, "checkpoints");
  }

  private void assertNonemptyFileWithPrefixExist(File parent, String prefix) {
    File[] files = parent.listFiles();
    assertNotNull(files);
    List<File> checkpointFiles = Arrays.stream(files)
        .filter(File::isFile)
        .filter(file -> file.getName().startsWith(prefix)).collect(toList());
    assertFalse(checkpointFiles.isEmpty());
  }

  private void blockMasterCheckpointUfsJournal() throws Exception {
    // Perform operations to generate a checkpoint.
    for (int i = 0; i < CHECKPOINT_SIZE * 2; i++) {
      FileOutStream out = mFs.createFile(new AlluxioURI("/" + i));
      out.write(new byte[1]);
      out.close();
    }
    IntegrationTestUtils.waitForUfsJournalCheckpoint(Constants.BLOCK_MASTER_NAME);
  }

  @Test
  @LocalAlluxioClusterResource.Config(confParams = {PropertyKey.Name.MASTER_JOURNAL_TYPE,
      "EMBEDDED", PropertyKey.Name.MASTER_METASTORE, "HEAP"})
  public void dumpHeapCheckpointFromEmbeddedJournal() throws Throwable {
    for (String name : Arrays.asList("/pin", "/max_replication", "/async_persist", "/ttl")) {
      mFs.createFile(new AlluxioURI(name)).close();
    }
    mFs.setAttribute(new AlluxioURI("/pin"),
        SetAttributePOptions.newBuilder().setPinned(true).build());
    mFs.setAttribute(new AlluxioURI("/max_replication"),
        SetAttributePOptions.newBuilder().setReplicationMax(5).build());
    mFs.persist(new AlluxioURI("/async_persist"));
    mFs.setAttribute(new AlluxioURI("/ttl"),
        SetAttributePOptions.newBuilder()
            .setCommonOptions(FileSystemMasterCommonPOptions.newBuilder().setTtl(100000).build())
            .build());
    checkpointEmbeddedJournal();

    mLocalAlluxioClusterResource.get().stopMasters();
    // Read current snapshot index.
    final long snapshotIdxTarget = CHECKPOINT_SIZE * 2 + 50;
    long currentSnapshotIdx = getCurrentRatisSnapshotIndex(
        mLocalAlluxioClusterResource.get().getLocalAlluxioMaster().getJournalFolder());
    // Verify it's beyond target.
    assertTrue("snapshot should be beyond target", currentSnapshotIdx >= snapshotIdxTarget);
    // Make sure to point the tool to leader's journal folder.
    String leaderJournalDir =
        mLocalAlluxioClusterResource.get().getLocalAlluxioMaster().getJournalFolder();
    // Run journal tool.
    JournalTool.main(
        new String[] {"-inputDir", leaderJournalDir, "-outputDir", mDumpDir.getAbsolutePath()});
    // Find the main checkpoint dir.
    String checkpointDir = findCheckpointDir();

    for (String subPath : Arrays.asList("HEAP_INODE_STORE", "INODE_COUNTER",
        "PINNED_INODE_FILE_IDS", "REPLICATION_LIMITED_FILE_IDS", "TO_BE_PERSISTED_FILE_IDS",
        "INODE_DIRECTORY_ID_GENERATOR")) {
      assertNonemptyFileExists(PathUtils.concatPath(checkpointDir, subPath));
    }
  }

  @Test
  @LocalAlluxioClusterResource.Config(confParams = {PropertyKey.Name.MASTER_METASTORE, "ROCKS"})
  public void dumpRocksCheckpointFromUfsJournal() throws Throwable {
    checkpointUfsJournal();
    JournalTool.main(new String[] {"-outputDir", mDumpDir.getAbsolutePath()});
    String checkpointDir = findCheckpointDir();
    assertNonemptyFileExists(
        PathUtils.concatPath(checkpointDir, "INODE_TREE", "CACHING_INODE_STORE"));
  }

  private void checkpointUfsJournal() throws Exception {
    // Perform operations to generate a checkpoint.
    for (int i = 0; i < CHECKPOINT_SIZE * 2; i++) {
      mFs.createFile(new AlluxioURI("/" + i)).close();
    }
    IntegrationTestUtils.waitForUfsJournalCheckpoint(Constants.FILE_SYSTEM_MASTER_NAME);
  }

  private void checkpointEmbeddedJournal() throws Throwable {
    // Perform operations before generating a checkpoint.
    for (int i = 0; i < CHECKPOINT_SIZE * 2; i++) {
      mFs.createFile(new AlluxioURI("/" + i)).close();
    }

    // Take snapshot on master.
    new RetryHandlingMetaMasterClient(MasterClientContext
        .newBuilder(ClientContext.create(Configuration.global()))
        .setMasterInquireClient(new SingleMasterInquireClient(
            mLocalAlluxioClusterResource.get().getLocalAlluxioMaster().getAddress()))
        .build()).checkpoint();
  }

  private long getCurrentRatisSnapshotIndex(String journalFolder) throws Throwable {
    try (RaftStorage storage = StorageImplUtils.newRaftStorage(
        new File(RaftJournalUtils.getRaftJournalDir(new File(journalFolder)),
            RaftJournalSystem.RAFT_GROUP_UUID.toString()),
          RaftServerConfigKeys.Log.CorruptionPolicy.getDefault(),
          RaftStorage.StartupOption.RECOVER,
          RaftServerConfigKeys.STORAGE_FREE_SPACE_MIN_DEFAULT.getSize())) {
      storage.initialize();
      SimpleStateMachineStorage stateMachineStorage = new SimpleStateMachineStorage();
      stateMachineStorage.init(storage);
      SingleFileSnapshotInfo snapshot = stateMachineStorage.getLatestSnapshot();
      if (snapshot == null) {
        throw new IOException("Failed to find a valid snapshot");
      }
      return snapshot.getIndex();
    }
  }

  private String findCheckpointDir() throws IOException {
    List<Path> checkpoint = Files.list(mDumpDir.toPath())
        .filter(p -> p.toString().contains("checkpoints-")).collect(toList());
    assertEquals("Unexpected checkpoint list: " + checkpoint, 1, checkpoint.size());
    return checkpoint.get(0).toString();
  }

  private void assertNonemptyFileExists(String s) {
    File f = new File(s);
    assertTrue(f.exists());
    assertTrue(f.isFile());
    assertThat(f.length(), Matchers.greaterThan(0L));
  }

  private void assertNonemptyDirExists(String s) {
    File f = new File(s);
    assertTrue(f.exists());
    assertTrue(f.isDirectory());
    assertThat(f.list().length, Matchers.greaterThan(0));
  }
}
