alluxio-master-2.x/core/server/master/src/test/java/alluxio/master/file/FileSystemMasterSyncMetadataFlushJournalTest.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.master.file;

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

import alluxio.AlluxioURI;
import alluxio.exception.AccessControlException;
import alluxio.exception.InvalidPathException;
import alluxio.exception.status.UnavailableException;
import alluxio.file.options.DescendantType;
import alluxio.grpc.FileSystemMasterCommonPOptions;
import alluxio.master.file.contexts.InternalOperationContext;
import alluxio.master.file.meta.InodeTree;
import alluxio.master.file.meta.LockingScheme;
import alluxio.master.file.meta.MutableInode;
import alluxio.master.file.meta.UfsSyncPathCache;
import alluxio.master.journal.FileSystemMergeJournalContext;
import alluxio.master.journal.JournalContext;
import alluxio.master.journal.MetadataSyncMergeJournalContext;
import alluxio.proto.journal.Journal;
import alluxio.underfs.UnderFileSystem;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mockito;
import org.powermock.core.classloader.annotations.PrepareForTest;
import org.powermock.modules.junit4.PowerMockRunner;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

@RunWith(PowerMockRunner.class)
@PrepareForTest({UnderFileSystem.Factory.class})
public class FileSystemMasterSyncMetadataFlushJournalTest
    extends FileSystemMasterSyncMetadataTestBase {
  public FileSystemMasterSyncMetadataFlushJournalTest() {
  }

  @Override
  public void before() throws Exception {
    super.before();

    // verify the files don't exist in alluxio
    assertEquals(1, mFileSystemMaster.getInodeTree().getInodeCount());
  }

  @Test
  public void hierarchicalDirectory() throws Exception {
    run(3, 5);
  }

  @Test
  public void flatDirectory() throws Exception {
    run(1, 100);
  }

  @Test
  public void runFailedHierarchical()
      throws IOException, AccessControlException, InvalidPathException {
    mUfs.mFailedPaths.clear();
    mUfs.mFailedPaths.add("0_1");

    cleanupUfs();
    createTestUfs(3, 5);

    TestInodeSyncStream iss;

    TestJournalContext testJournalContext = new TestJournalContext();
    try (JournalContext journalContext = Mockito.spy(new FileSystemMergeJournalContext(
        testJournalContext,
        new FileSystemJournalEntryMerger()))) {
      iss = makeInodeSyncStream("/", journalContext);
      assertEquals(iss.sync(), InodeSyncStream.SyncStatus.FAILED);
      iss.assertAllJournalFlushedIntoAsyncJournalWriter();
    }
  }

  @Test
  public void runFailedFlat()
      throws IOException {
    mUfs.mFailedPaths.clear();
    mUfs.mFailedPaths.add("/");

    cleanupUfs();
    createTestUfs(1, 100);

    AtomicReference<TestInodeSyncStream> iss = new AtomicReference<>();

    TestJournalContext testJournalContext = new TestJournalContext();
    assertThrows(RuntimeException.class, () -> {
      try (JournalContext journalContext = Mockito.spy(new FileSystemMergeJournalContext(
          testJournalContext,
          new FileSystemJournalEntryMerger()))) {
        iss.set(makeInodeSyncStream("/", journalContext));
        assertEquals(iss.get().sync(), InodeSyncStream.SyncStatus.FAILED);
        iss.get().assertAllJournalFlushedIntoAsyncJournalWriter();
      }
    });
  }

  private void run(int numLevels, int numInodesPerLevel)
      throws Exception {
    int current = numInodesPerLevel;
    int sum = 0;
    for (int i = 0; i < numLevels; ++i) {
      sum += current;
      current *= numInodesPerLevel;
    }
    int numExpectedInodes = sum + 1;
    int numExpectedDirectories = sum - current / numInodesPerLevel;
    int numExpectedFiles = current / numInodesPerLevel;
    cleanupUfs();
    createTestUfs(numLevels, numInodesPerLevel);

    // Test load metadata from UFS
    TestInodeSyncStream iss;
    TestJournalContext testJournalContext = new TestJournalContext();
    try (JournalContext journalContext = Mockito.spy(new FileSystemMergeJournalContext(
        testJournalContext,
        new FileSystemJournalEntryMerger()))) {
      iss = makeInodeSyncStream("/", journalContext);
      assertEquals(iss.sync(), InodeSyncStream.SyncStatus.OK);
      // No journal should be written once the metadata sync is done.
      testJournalContext.mAllowAppendingOrFlushingJournals = false;
      iss.assertAllJournalFlushedIntoAsyncJournalWriter();
    }
    // A. 1 journal entry for an inode file
    // B. 3-4 journals for a directory
    //    a. generating inode directory id
    //    b. create inode file
    //    c. set children loaded
    //    d. (maybe) update access time
    // C. 1 journals for root
    //    a. set children loaded
    // D. up to numInodePerLevel journals for updating root access time
    assertEquals(numExpectedInodes, mFileSystemMaster.getInodeTree().getInodeCount());
    assertTrue(testJournalContext.mAppendedEntries.size()
        >= numExpectedFiles + numExpectedDirectories * 3);
    assertTrue(testJournalContext.mAppendedEntries.size()
        <= numExpectedFiles + numExpectedDirectories * 4 + 1 + numExpectedInodes);
    Set<MutableInode<?>> inodes = mFileSystemMaster.getInodeStore().allInodes();
    for (MutableInode<?> inode: inodes) {
      if (inode.isDirectory()) {
        assertTrue(inode.asDirectory().isDirectChildrenLoaded());
      }
      if (inode.isFile()) {
        assertTrue(inode.asFile().isCompleted());
      }
    }
    assertEquals(1, testJournalContext.mFlushCount.get());

    // Update inode metadata in UFS and the metadata sync should delete inodes and then create them
    // sleep 1 sec to update the timestamps for files in UFS
    Thread.sleep(1000);
    cleanupUfs();
    createUfsHierarchy(0, numLevels, "", numInodesPerLevel);
    testJournalContext = new TestJournalContext();
    try (JournalContext journalContext = Mockito.spy(new FileSystemMergeJournalContext(
        testJournalContext,
        new FileSystemJournalEntryMerger()))) {
      iss = makeInodeSyncStream("/", journalContext);
      assertEquals(iss.sync(), InodeSyncStream.SyncStatus.OK);
      // No journal should be written once the metadata sync is done.
      testJournalContext.mAllowAppendingOrFlushingJournals = false;
      iss.assertAllJournalFlushedIntoAsyncJournalWriter();
    }
    assertEquals(numExpectedInodes, mFileSystemMaster.getInodeTree().getInodeCount());
    assertEquals(numExpectedFiles * 2, testJournalContext.mAppendedEntries.size());
    assertEquals(1, testJournalContext.mFlushCount.get());

    // Delete inodes in UFS and trigger a metadata sync
    cleanupUfs();
    testJournalContext = new TestJournalContext();
    try (JournalContext journalContext = Mockito.spy(new FileSystemMergeJournalContext(
        testJournalContext,
        new FileSystemJournalEntryMerger()))) {
      iss = makeInodeSyncStream("/", journalContext);
      assertEquals(iss.sync(), InodeSyncStream.SyncStatus.OK);
      // No journal should be written once the metadata sync is done.
      testJournalContext.mAllowAppendingOrFlushingJournals = false;
      iss.assertAllJournalFlushedIntoAsyncJournalWriter();
    }
    assertEquals(1, mFileSystemMaster.getInodeTree().getInodeCount());
    assertEquals(numExpectedInodes - 1, testJournalContext.mAppendedEntries.size());
    assertEquals(1, mFileSystemMaster.getInodeCount());
    assertEquals(1, testJournalContext.mFlushCount.get());
  }

  private TestInodeSyncStream makeInodeSyncStream(String path, JournalContext journalContext) {
    FileSystemMasterCommonPOptions options = FileSystemMasterCommonPOptions.newBuilder()
        .setSyncIntervalMs(0)
        .build();
    DescendantType descendantType = DescendantType.ALL;
    try {
      LockingScheme syncScheme =
          new LockingScheme(new AlluxioURI(path), InodeTree.LockPattern.READ, options,
              mFileSystemMaster.getSyncPathCache(), descendantType); // shouldSync
      return
          new TestInodeSyncStream(
              syncScheme, mFileSystemMaster, mFileSystemMaster.getSyncPathCache(),
              new RpcContext(NoopBlockDeletionContext.INSTANCE,
                  journalContext, new InternalOperationContext()), descendantType, options,
              false,
              false,
              false);
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
  }

  private void createTestUfs(int numLevels, int numInodesPerLevel) throws IOException {
    createUfsHierarchy(0, numLevels, "", numInodesPerLevel);
  }

  private static class TestInodeSyncStream extends InodeSyncStream {
    private List<MetadataSyncMergeJournalContext> mJournalContexts = new ArrayList<>();

    public TestInodeSyncStream(
        LockingScheme rootScheme, DefaultFileSystemMaster fsMaster,
        UfsSyncPathCache syncPathCache, RpcContext rpcContext,
        DescendantType descendantType,
        FileSystemMasterCommonPOptions options,
        boolean forceSync, boolean loadOnly, boolean loadAlways) {
      super(rootScheme, fsMaster, syncPathCache, rpcContext, descendantType, options, forceSync,
          loadOnly, loadAlways);
    }

    @Override
    protected synchronized RpcContext getMetadataSyncRpcContext() {
      RpcContext context = super.getMetadataSyncRpcContext();
      if (context.getJournalContext() instanceof MetadataSyncMergeJournalContext) {
        mJournalContexts.add((MetadataSyncMergeJournalContext) context.getJournalContext());
      }
      return context;
    }

    public void assertAllJournalFlushedIntoAsyncJournalWriter() {
      for (MetadataSyncMergeJournalContext journalContext: mJournalContexts) {
        assertEquals(0, journalContext.getMerger().getMergedJournalEntries().size());
      }
    }
  }

  private static class TestJournalContext implements JournalContext {
    List<Journal.JournalEntry> mAppendedEntries = Collections.synchronizedList(new ArrayList<>());
    List<Journal.JournalEntry> mPendingEntries = Collections.synchronizedList(new ArrayList<>());
    AtomicInteger mFlushCount = new AtomicInteger();
    volatile boolean mAllowAppendingOrFlushingJournals = true;

    @Override
    public synchronized void append(Journal.JournalEntry entry) {
      assertTrue(mAllowAppendingOrFlushingJournals);
      mAppendedEntries.add(entry);
      mPendingEntries.add(entry);
    }

    @Override
    public synchronized void flush() throws UnavailableException {
      if (mPendingEntries.size() > 0) {
        mPendingEntries.clear();
        mFlushCount.incrementAndGet();
      }
    }

    @Override
    public synchronized void close() throws UnavailableException {
      flush();
    }
  }
}

