TestFuseDFS.java

/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import java.io.*;
import java.net.URI;
import java.util.ArrayList;
import java.util.concurrent.atomic.*;

import org.slf4j.event.Level;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.fs.permission.*;
import org.apache.hadoop.hdfs.*;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.test.GenericTestUtils;
import org.apache.hadoop.util.StringUtils;

import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.AfterAll;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.fail;

/**
 * Basic functional tests on a fuse-dfs mount.
 */
public class TestFuseDFS {

  private static MiniDFSCluster cluster;
  private static FileSystem fs;
  private static Process fuseProcess;
  private static Runtime r;
  private static String mountPoint;

  private static final Logger LOG = LoggerFactory.getLogger(TestFuseDFS.class);
  {
    GenericTestUtils.setLogLevel(LOG, Level.TRACE);
  }

  /** Dump the given intput stream to stderr */
  private static void dumpInputStream(InputStream is) throws IOException {
    int len;
    do {
      byte b[] = new byte[is.available()];
      len = is.read(b);
      System.out.println("Read "+len+" bytes");
      System.out.write(b, 0, b.length);
    } while (len > 0);
  }

  /** 
   * Wait for the given process to return and check that it exited
   * as required. Log if the process failed.
   */
  private static void checkProcessRet(Process p, boolean expectPass) 
      throws IOException {
    try {
      int ret = p.waitFor();
      if (ret != 0) {
	dumpInputStream(p.getErrorStream());
      }
      if (expectPass) {
	assertEquals(0, ret);
      } else {
	assertTrue(ret != 0);
      }
    } catch (InterruptedException ie) {
      fail("Process interrupted: "+ie.getMessage());
    }
  }

  /** Exec the given command and assert it executed successfully */
  private static void execWaitRet(String cmd) throws IOException {
    LOG.debug("EXEC "+cmd);
    Process p = r.exec(cmd);
    try {
      p.waitFor();
    } catch (InterruptedException ie) {
      fail("Process interrupted: "+ie.getMessage());
    }
  }

  /** Exec the given command and assert it executed successfully */
  private static void execIgnoreRet(String cmd) throws IOException {
    LOG.debug("EXEC "+cmd);
    r.exec(cmd);
  }

  /** Exec the given command and assert it executed successfully */
  private static void execAssertSucceeds(String cmd) throws IOException {
    LOG.debug("EXEC "+cmd);
    checkProcessRet(r.exec(cmd), true);
  }

  /** Exec the given command, assert it returned an error code */
  private static void execAssertFails(String cmd) throws IOException {
    LOG.debug("EXEC "+cmd);
    checkProcessRet(r.exec(cmd), false);
  }

  /** Create and write the given file */
  private static void createFile(File f, String s) throws IOException {
    InputStream is = new ByteArrayInputStream(s.getBytes());
    FileOutputStream fos = new FileOutputStream(f);
    IOUtils.copyBytes(is, fos, s.length(), true);
  }

  /** Check that the given file exists with the given contents */
  private static void checkFile(File f, String expectedContents) 
      throws IOException {
    FileInputStream fi = new FileInputStream(f);
    int len = expectedContents.length();
    byte[] b = new byte[len];
    try {
      IOUtils.readFully(fi, b, 0, len);
    } catch (IOException ie) {
      fail("Reading "+f.getName()+" failed with "+ie.getMessage());
    } finally {
      fi.close(); // NB: leaving f unclosed prevents unmount
    }
    String s = new String(b, 0, len);
    assertEquals(expectedContents, s, "File content differs");
  }

  private static class RedirectToStdoutThread extends Thread {
    private InputStream is;

    RedirectToStdoutThread(InputStream is) {
      this.is = is;
    }
    public void run() {
      try {
        InputStreamReader isr = new InputStreamReader(is);
        BufferedReader br = new BufferedReader(isr);
        String line=null;
        while ( (line = br.readLine()) != null) {
          LOG.error("FUSE_LINE:" + line);
        }
      } catch (IOException e) {
        e.printStackTrace();
      }
    }
  }

  /** Run a fuse-dfs process to mount the given DFS */
  private static Process establishMount(URI uri) throws IOException  {
    Runtime r = Runtime.getRuntime();
    String cp = System.getProperty("java.class.path");

    String buildTestDir = System.getProperty("build.test");
    String fuseCmd = buildTestDir + "/../fuse_dfs";
    String libHdfs = buildTestDir + "/../../../c++/lib";

    String arch = System.getProperty("os.arch");
    String jvm = System.getProperty("java.home") + "/lib/" + arch + "/server";
    String lp = System.getProperty("LD_LIBRARY_PATH")+":"+libHdfs+":"+jvm;
    LOG.debug("LD_LIBRARY_PATH=" + lp);

    String nameNode = 
      "dfs://" + uri.getHost() + ":" + String.valueOf(uri.getPort());

    // NB: We're mounting via an unprivileged user, therefore
    // user_allow_other needs to be set in /etc/fuse.conf, which also
    // needs to be world readable.
    String mountCmd[] = {
      fuseCmd, nameNode, mountPoint,
      // "-odebug",              // Don't daemonize
      "-obig_writes",            // Allow >4kb writes
      "-oentry_timeout=0.1",     // Don't cache dents long
      "-oattribute_timeout=0.1", // Don't cache attributes long
      "-ononempty",              // Don't complain about junk in mount point
      "-f",                      // Don't background the process
      "-ordbuffer=32768",        // Read buffer size in kb
      "-omax_background=100",    // Set fuse max_background=100 (12 by default)
      "rw"
    };

    String [] env = {
      "CLASSPATH="+cp,
      "LD_LIBRARY_PATH="+lp,
      "PATH=/usr/bin:/bin"
    };

    execWaitRet("fusermount -u " + mountPoint);
    execAssertSucceeds("rm -rf " + mountPoint);
    execAssertSucceeds("mkdir -p " + mountPoint);

    // Mount the mini cluster
    String cmdStr = "";
    for (String c : mountCmd) {
      cmdStr += (" " + c);
    }
    LOG.info("now mounting with:" + cmdStr);
    Process fuseProcess = r.exec(mountCmd, env);
    RedirectToStdoutThread stdoutThread =
      new RedirectToStdoutThread(fuseProcess.getInputStream());
    RedirectToStdoutThread stderrThread =
      new RedirectToStdoutThread(fuseProcess.getErrorStream());
    stdoutThread.start();
    stderrThread.start();
    // Wait for fusermount to start up, so that we know we're operating on the
    // FUSE FS when we run the tests.
    try {
      Thread.sleep(50000);
    } catch (InterruptedException e) {
    }
    return fuseProcess;
  }

  /** Tear down the fuse-dfs process and mount */
  private static void teardownMount() throws IOException {
    execWaitRet("fusermount -u " + mountPoint);
    try {
      assertEquals(0,
          fuseProcess.waitFor()); // fuse_dfs should exit cleanly
    } catch (InterruptedException e) {
      fail("interrupted while waiting for fuse_dfs process to exit.");
    }
  }

  @BeforeAll
  public static void startUp() throws IOException {
    Configuration conf = new HdfsConfiguration();
    r = Runtime.getRuntime();
    mountPoint = System.getProperty("build.test") + "/mnt";
    conf.setBoolean(DFSConfigKeys.DFS_PERMISSIONS_ENABLED_KEY, false);
    cluster = new MiniDFSCluster.Builder(conf).build();
    cluster.waitClusterUp();
    fs = cluster.getFileSystem();
    fuseProcess = establishMount(fs.getUri());
  }

  @AfterAll
  public static void tearDown() throws IOException {
    // Unmount before taking down the mini cluster
    // so no outstanding operations hang.
    teardownMount();
    if (fs != null) {
      fs.close();
    }
    if (cluster != null) {
      cluster.shutdown();
    }
  }

  /** Test basic directory creation, access, removal */
  @Test
  public void testBasicDir() throws IOException {
    File d = new File(mountPoint, "dir1");

    // Mkdir, access and rm via the mount
    execAssertSucceeds("mkdir " + d.getAbsolutePath());
    execAssertSucceeds("ls " + d.getAbsolutePath());
    execAssertSucceeds("rmdir " + d.getAbsolutePath());

    // The dir should no longer exist
    execAssertFails("ls " + d.getAbsolutePath());
  }

  /** Test basic file creation and writing */
  @Test
  public void testCreate() throws IOException {
    final String contents = "hello world";
    File f = new File(mountPoint, "file1");

    // Create and access via the mount
    createFile(f, contents);

    // XX avoids premature EOF
    try {
      Thread.sleep(1000);
    } catch (InterruptedException ie) { }

    checkFile(f, contents);

    // Cat, stat and delete via the mount
    execAssertSucceeds("cat " + f.getAbsolutePath());
    execAssertSucceeds("stat " + f.getAbsolutePath());
    execAssertSucceeds("rm " + f.getAbsolutePath());

    // The file should no longer exist
    execAssertFails("ls " + f.getAbsolutePath());
  }

  /** Test creating a file via touch */
  @Test
  public void testTouch() throws IOException {
    File f = new File(mountPoint, "file1");
    execAssertSucceeds("touch " + f.getAbsolutePath());
    execAssertSucceeds("rm " + f.getAbsolutePath());
  }

  /** Test random access to a file */
  @Test
  public void testRandomAccess() throws IOException {
    final String contents = "hello world";
    File f = new File(mountPoint, "file1");

    createFile(f, contents);

    RandomAccessFile raf = new RandomAccessFile(f, "rw");
    raf.seek(f.length());
    try {
      raf.write('b');
    } catch (IOException e) {
      // Expected: fuse-dfs not yet support append
      assertEquals("Operation not supported", e.getMessage());
    } finally {
      raf.close();
    }

    raf = new RandomAccessFile(f, "rw");
    raf.seek(0);
    try {
      raf.write('b');
      fail("Over-wrote existing bytes");
    } catch (IOException e) {
      // Expected: can-not overwrite a file
      assertEquals("Invalid argument", e.getMessage());
    } finally {
      raf.close();
    }
    execAssertSucceeds("rm " + f.getAbsolutePath());
  }

  /** Test copying a set of files from the mount to itself */
  @Test
  public void testCopyFiles() throws IOException {
    final String contents = "hello world";
    File d1 = new File(mountPoint, "dir1");
    File d2 = new File(mountPoint, "dir2");

    // Create and populate dir1 via the mount
    execAssertSucceeds("mkdir " + d1.getAbsolutePath());
    for (int i = 0; i < 5; i++) {
      createFile(new File(d1, "file"+i), contents);
    }
    assertEquals(5, d1.listFiles().length);

    // Copy dir from the mount to the mount
    execAssertSucceeds("cp -r " + d1.getAbsolutePath() +
                       " " + d2.getAbsolutePath());
    assertEquals(5, d2.listFiles().length);

    // Access all the files in the dirs and remove them
    execAssertSucceeds("find " + d1.getAbsolutePath());
    execAssertSucceeds("find " + d2.getAbsolutePath());
    execAssertSucceeds("rm -r " + d1.getAbsolutePath());
    execAssertSucceeds("rm -r " + d2.getAbsolutePath());
  }

  /** Test concurrent creation and access of the mount */
  @Test
  public void testMultipleThreads() throws IOException {
    ArrayList<Thread> threads = new ArrayList<Thread>();
    final AtomicReference<String> errorMessage = new AtomicReference<String>();

    for (int i = 0; i < 10; i++) {
      Thread t = new Thread() {
	  public void run() {
	    try {
	      File d = new File(mountPoint, "dir"+getId());
	      execWaitRet("mkdir " + d.getAbsolutePath());
	      for (int j = 0; j < 10; j++) {
		File f = new File(d, "file"+j);
		final String contents = "thread "+getId()+" "+j;
		createFile(f, contents);
	      }
	      for (int j = 0; j < 10; j++) {
		File f = new File(d, "file"+j);
		execWaitRet("cat " + f.getAbsolutePath());
		execWaitRet("rm " + f.getAbsolutePath());
	      }
	      execWaitRet("rmdir " + d.getAbsolutePath());
	    } catch (IOException ie) {
	      errorMessage.set(
		String.format("Exception %s", 
			      StringUtils.stringifyException(ie)));
	    }
          }
	};
      t.start();
      threads.add(t);
    }

    for (Thread t : threads) {
      try {
	t.join();
      } catch (InterruptedException ie) {
	fail("Thread interrupted: "+ie.getMessage());
      }
    }

    assertNull(errorMessage.get(), errorMessage.get());
  }
}