HDFS-729. NameNode API to list files that have missing blocks.

(Rodrigo Schmidt via dhruba)



git-svn-id: https://svn.apache.org/repos/asf/hadoop/hdfs/trunk@920713 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -89,6 +89,9 @@ Trunk (unreleased changes)
 
     HDFS-998. Quote blocks streamed through jsps. (cdouglas)
 
+    HDFS-729. NameNode API to list files that have missing blocks.
+    (Rodrigo Schmidt via dhruba)
+
   OPTIMIZATIONS
 
     HDFS-946. NameNode should not return full path name when lisitng a

src/java/org/apache/hadoop/hdfs/protocol/ClientProtocol.java

@@ -22,6 +22,7 @@ import java.io.IOException;
 
 import org.apache.hadoop.fs.ContentSummary;
 import org.apache.hadoop.fs.CreateFlag;
+import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FsServerDefaults;
 import org.apache.hadoop.fs.Options;
 import org.apache.hadoop.fs.UnresolvedLinkException;
@@ -54,9 +55,9 @@ public interface ClientProtocol extends VersionedProtocol {
    * Compared to the previous version the following changes have been introduced:
    * (Only the latest change is reflected.
    * The log of historical changes can be retrieved from the svn).
-   * 58: Add symlink APIs.
+   * 59: Add API to retrive corrupted block list
    */
-  public static final long versionID = 58L;
+  public static final long versionID = 59L;
   
   ///////////////////////////////////////
   // File contents
@@ -530,6 +531,15 @@ public interface ClientProtocol extends VersionedProtocol {
    */
   public void metaSave(String filename) throws IOException;
 
+  /**
+   * @return Array of FileStatus objects referring to corrupted files.
+   *         The server could return all or a few of the files that are corrupt.
+   * @throws AccessControlException
+   * @throws IOException
+   */
+  FileStatus[] getCorruptFiles() 
+    throws AccessControlException, IOException; 
+  
   /**
    * Get the file info for a specific file or directory.
    * @param src The string representation of the path to the file

src/java/org/apache/hadoop/hdfs/server/namenode/BlockManager.java

@@ -24,6 +24,7 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.Iterator;
+import java.util.LinkedHashSet;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
@@ -51,6 +52,7 @@ public class BlockManager {
   // Default initial capacity and load factor of map
   public static final int DEFAULT_INITIAL_MAP_CAPACITY = 16;
   public static final float DEFAULT_MAP_LOAD_FACTOR = 0.75f;
+  public static final int DEFAULT_MAX_CORRUPT_FILES_RETURNED = 500;
 
   private final FSNamesystem namesystem;
 
@@ -105,7 +107,9 @@ public class BlockManager {
   int minReplication;
   // Default number of replicas
   int defaultReplication;
-
+  // How many entries are returned by getCorruptInodes()
+  int maxCorruptFilesReturned;
+  
   // variable to enable check for enough racks 
   boolean shouldCheckForEnoughRacks = true;
 
@@ -140,6 +144,8 @@ public class BlockManager {
                          namesystem,
                          namesystem.clusterMap);
 
+    this.maxCorruptFilesReturned = conf.getInt("dfs.corruptfilesreturned.max",
+        DEFAULT_MAX_CORRUPT_FILES_RETURNED);
     this.defaultReplication = conf.getInt("dfs.replication", 3);
     this.maxReplication = conf.getInt("dfs.replication.max", 512);
     this.minReplication = conf.getInt(DFSConfigKeys.DFS_NAMENODE_REPLICATION_MIN_KEY,
@@ -1706,4 +1712,25 @@ public class BlockManager {
                                                      startingBlockId);
   }  
   
+  /**
+   * @return inodes of files with corrupt blocks, with a maximum of 
+   * MAX_CORRUPT_FILES_RETURNED inodes listed in total
+   */
+  INode[] getCorruptInodes() {
+    LinkedHashSet<INode> set = new LinkedHashSet<INode>();
+
+    for (Block blk : 
+            neededReplications.getQueue(
+                UnderReplicatedBlocks.QUEUE_WITH_CORRUPT_BLOCKS)){
+      INode inode = blocksMap.getINode(blk);
+      if (inode != null && countNodes(blk).liveReplicas() == 0) {
+        set.add(inode);
+        if (set.size() >= this.maxCorruptFilesReturned) {
+          break;  
+        }
+      } 
+    }
+    return set.toArray(new INode[set.size()]);
+  }
+  
 }

src/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java

@@ -63,6 +63,7 @@ import org.apache.hadoop.hdfs.DFSConfigKeys;
 import org.apache.hadoop.fs.ContentSummary;
 import org.apache.hadoop.fs.CreateFlag;
 import org.apache.hadoop.fs.FileAlreadyExistsException;
+import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FsServerDefaults;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.UnresolvedLinkException;
@@ -4387,6 +4388,38 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean, FSClusterSt
                                                   startingBlockId);
   }
 
+  /**
+   * @return Array of FileStatus objects representing files with 
+   * corrupted blocks.
+   * @throws AccessControlException
+   * @throws IOException
+   */
+  synchronized FileStatus[] getCorruptFiles() 
+    throws AccessControlException, IOException {
+    
+    checkSuperuserPrivilege();
+    
+    INode[] inodes = blockManager.getCorruptInodes();
+    FileStatus[] ret = new FileStatus[inodes.length];
+    
+    int i = 0;
+    for (INode inode: inodes) {
+      String src = inode.getFullPathName();
+      ret[i++] = new FileStatus(inode.computeContentSummary().getLength(), 
+          inode.isDirectory(), 
+          ((INodeFile)inode).getReplication(), 
+          ((INodeFile)inode).getPreferredBlockSize(),
+          inode.getModificationTime(),
+          inode.getAccessTime(),
+          inode.getFsPermission(),
+          inode.getUserName(),
+          inode.getGroupName(),
+          new Path(src));
+    }
+
+    return ret;
+  }
+  
   public synchronized ArrayList<DatanodeDescriptor> getDecommissioningNodes() {
     ArrayList<DatanodeDescriptor> decommissioningNodes = 
         new ArrayList<DatanodeDescriptor>();

src/java/org/apache/hadoop/hdfs/server/namenode/NameNode.java

@@ -31,6 +31,7 @@ import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.ContentSummary;
 import org.apache.hadoop.fs.CreateFlag;
+import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.FsServerDefaults;
 import org.apache.hadoop.fs.Path;
@@ -968,6 +969,14 @@ public class NameNode implements ClientProtocol, DatanodeProtocol,
     namesystem.metaSave(filename);
   }
 
+  /** {@inheritDoc} */
+  public FileStatus[] getCorruptFiles() 
+    throws AccessControlException, IOException {
+    
+    return namesystem.getCorruptFiles();
+    
+  }
+  
   /** {@inheritDoc} */
   public ContentSummary getContentSummary(String path) throws IOException {
     return namesystem.getContentSummary(path);

src/java/org/apache/hadoop/hdfs/server/namenode/UnderReplicatedBlocks.java

@@ -27,6 +27,7 @@ import org.apache.hadoop.hdfs.protocol.Block;
  */
 class UnderReplicatedBlocks implements Iterable<Block> {
   static final int LEVEL = 4;
+  static public final int QUEUE_WITH_CORRUPT_BLOCKS = 2;
   private List<TreeSet<Block>> priorityQueues = new ArrayList<TreeSet<Block>>();
       
   /* constructor */
@@ -81,7 +82,7 @@ class UnderReplicatedBlocks implements Iterable<Block> {
       if (decommissionedReplicas > 0) {
         return 0;
       }
-      return 2; // keep these blocks in needed replication.
+      return QUEUE_WITH_CORRUPT_BLOCKS; // keep these blocks in needed replication.
     } else if(curReplicas==1) {
       return 0; // highest priority
     } else if(curReplicas*3<expectedReplicas) {
@@ -183,7 +184,15 @@ class UnderReplicatedBlocks implements Iterable<Block> {
                                     + " at priority level " + curPri);
     }
   }
-      
+
+  /* returns an interator of all blocks in a given priority queue */
+  public synchronized Iterable<Block> getQueue(int priority) {
+    if (priority < 0 || priority >= LEVEL) {
+      return null;
+    }
+    return priorityQueues.get(priority);
+  }
+  
   /* return an iterator of all the under replication blocks */
   public synchronized BlockIterator iterator() {
     return new BlockIterator();

src/test/hdfs/org/apache/hadoop/hdfs/TestDFSClientRetries.java

@@ -260,6 +260,11 @@ public class TestDFSClientRetries extends TestCase {
 
     public void setTimes(String src, long mtime, long atime) throws IOException {}
 
+    public FileStatus[] getCorruptFiles()
+      throws AccessControlException, IOException {
+      return null;
+    }
+
     public void createSymlink(String target, String newpath, 
         FsPermission dirPerm, boolean createPath) 
         throws IOException, UnresolvedLinkException {}

src/test/hdfs/org/apache/hadoop/hdfs/TestFileCorruption.java

@@ -22,23 +22,43 @@ import java.io.DataInputStream;
 import java.io.DataOutputStream;
 import java.io.File;
 import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
 import java.util.ArrayList;
+import java.util.Random;
 
 import junit.framework.TestCase;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.impl.Log4JLogger;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.ChecksumException;
+import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hdfs.protocol.Block;
+import org.apache.hadoop.hdfs.protocol.ClientProtocol;
 import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
 import org.apache.hadoop.hdfs.server.common.GenerationStamp;
 import org.apache.hadoop.hdfs.server.datanode.DataNode;
+import org.apache.hadoop.hdfs.server.namenode.FSNamesystem;
+import org.apache.hadoop.hdfs.server.namenode.NameNode;
+import org.apache.log4j.Level;
 
 /**
  * A JUnit test for corrupted file handling.
  */
 public class TestFileCorruption extends TestCase {
+  {
+    ((Log4JLogger)NameNode.stateChangeLog).getLogger().setLevel(Level.ALL);
+    ((Log4JLogger)FSNamesystem.LOG).getLogger().setLevel(Level.ALL);
+    ((Log4JLogger)DFSClient.LOG).getLogger().setLevel(Level.ALL);
+    ((Log4JLogger)DataNode.LOG).getLogger().setLevel(Level.ALL);
+  }
+  static Log LOG = ((Log4JLogger)NameNode.stateChangeLog);
+
   /** check if DFS can handle corrupted blocks properly */
   public void testFileCorruption() throws Exception {
     MiniDFSCluster cluster = null;
@@ -168,4 +188,172 @@ public class TestFileCorruption extends TestCase {
     }
     return new Block(blockId, blocks[idx].length(), blockTimeStamp);
   }
+
+  /** check if ClientProtocol.getCorruptFiles() returns a file that has missing blocks */
+  public void testCorruptFilesMissingBlock() throws Exception {
+    MiniDFSCluster cluster = null;
+    try {
+      Configuration conf = new HdfsConfiguration();
+      conf.setInt("dfs.datanode.directoryscan.interval", 1); // datanode scans directories
+      conf.setInt("dfs.blockreport.intervalMsec", 3 * 1000); // datanode sends block reports
+      cluster = new MiniDFSCluster(conf, 1, true, null);
+      FileSystem fs = cluster.getFileSystem();
+
+      // create two files with one block each
+      DFSTestUtil util = new DFSTestUtil("testCorruptFilesMissingBlock", 2, 1, 512);
+      util.createFiles(fs, "/srcdat");
+
+      // verify that there are no bad blocks.
+      ClientProtocol namenode = DFSClient.createNamenode(conf);
+      FileStatus[] badFiles = namenode.getCorruptFiles();
+      assertTrue("Namenode has " + badFiles.length + " corrupt files. Expecting none.",
+          badFiles.length == 0);
+
+      // Now deliberately remove one block
+      File data_dir = new File(System.getProperty("test.build.data"),
+      "dfs/data/data1/current/finalized");
+      assertTrue("data directory does not exist", data_dir.exists());
+      File[] blocks = data_dir.listFiles();
+      assertTrue("Blocks do not exist in data-dir", (blocks != null) && (blocks.length > 0));
+      for (int idx = 0; idx < blocks.length; idx++) {
+        if (!blocks[idx].getName().startsWith("blk_")) {
+          continue;
+        }
+        LOG.info("Deliberately removing file "+blocks[idx].getName());
+        assertTrue("Cannot remove file.", blocks[idx].delete());
+        break;
+      }
+
+      badFiles = namenode.getCorruptFiles();
+      while (badFiles.length == 0) {
+        Thread.sleep(1000);
+        badFiles = namenode.getCorruptFiles();
+      }
+      LOG.info("Namenode has bad files. " + badFiles.length);
+      assertTrue("Namenode has " + badFiles.length + " bad files. Expecting 1.",
+          badFiles.length == 1);
+      util.cleanup(fs, "/srcdat");
+    } finally {
+      if (cluster != null) { cluster.shutdown(); }
+    }
+  }
+
+  /** check if ClientProtocol.getCorruptFiles() returns the right limit */
+  public void testMaxCorruptFiles() throws Exception {
+    MiniDFSCluster cluster = null;
+    try {
+      Configuration conf = new HdfsConfiguration();
+      conf.setInt("dfs.corruptfilesreturned.max", 2);
+      conf.setInt("dfs.datanode.directoryscan.interval", 1); // datanode scans directories
+      conf.setInt("dfs.blockreport.intervalMsec", 3 * 1000); // datanode sends block reports
+      cluster = new MiniDFSCluster(conf, 1, true, null);
+      FileSystem fs = cluster.getFileSystem();
+
+      // create two files with one block each
+      DFSTestUtil util = new DFSTestUtil("testMaxCorruptFiles", 4, 1, 512);
+      util.createFiles(fs, "/srcdat2");
+
+      // verify that there are no bad blocks.
+      ClientProtocol namenode = DFSClient.createNamenode(conf);
+      FileStatus[] badFiles = namenode.getCorruptFiles();
+      assertTrue("Namenode has " + badFiles.length + " corrupt files. Expecting none.",
+          badFiles.length == 0);
+
+      // Now deliberately remove one block
+      File data_dir = new File(System.getProperty("test.build.data"),
+      "dfs/data/data1/current/finalized");
+      assertTrue("data directory does not exist", data_dir.exists());
+      File[] blocks = data_dir.listFiles();
+      assertTrue("Blocks do not exist in data-dir", (blocks != null) && (blocks.length > 0));
+      for (int idx = 0; idx < blocks.length; idx++) {
+        if (!blocks[idx].getName().startsWith("blk_")) {
+          continue;
+        }
+        LOG.info("Deliberately removing file "+blocks[idx].getName());
+        assertTrue("Cannot remove file.", blocks[idx].delete());
+      }
+
+      badFiles = namenode.getCorruptFiles();
+      while (badFiles.length < 2) {
+        badFiles = namenode.getCorruptFiles();
+        Thread.sleep(10000);
+      }
+      badFiles = namenode.getCorruptFiles(); // once more since time has passed
+      LOG.info("Namenode has bad files. " + badFiles.length);
+      assertTrue("Namenode has " + badFiles.length + " bad files. Expecting 2.",
+          badFiles.length == 2);
+      util.cleanup(fs, "/srcdat2");
+    } finally {
+      if (cluster != null) { cluster.shutdown(); }
+    }
+  }
+
+  /** check if ClientProtocol.getCorruptFiles() returns a file that has corrupted blocks */
+  public void testCorruptFilesCorruptedBlock() throws Exception {
+    MiniDFSCluster cluster = null;
+    Random random = new Random();
+    
+    try {
+      Configuration conf = new HdfsConfiguration();
+      conf.setInt("dfs.datanode.directoryscan.interval", 1); // datanode scans directories
+      conf.setInt("dfs.blockreport.intervalMsec", 3 * 1000); // datanode sends block reports
+      cluster = new MiniDFSCluster(conf, 1, true, null);
+      FileSystem fs = cluster.getFileSystem();
+
+      // create two files with one block each
+      DFSTestUtil util = new DFSTestUtil("testCorruptFilesCorruptedBlock", 2, 1, 512);
+      util.createFiles(fs, "/srcdat10");
+
+      // fetch bad file list from namenode. There should be none.
+      ClientProtocol namenode = DFSClient.createNamenode(conf);
+      FileStatus[] badFiles = namenode.getCorruptFiles();
+      assertTrue("Namenode has " + badFiles.length + " corrupt files. Expecting None.",
+          badFiles.length == 0);
+
+      // Now deliberately corrupt one block
+      File data_dir = new File(System.getProperty("test.build.data"),
+      "dfs/data/data1/current/finalized");
+      assertTrue("data directory does not exist", data_dir.exists());
+      File[] blocks = data_dir.listFiles();
+      assertTrue("Blocks do not exist in data-dir", (blocks != null) && (blocks.length > 0));
+      for (int idx = 0; idx < blocks.length; idx++) {
+        if (blocks[idx].getName().startsWith("blk_") &&
+            blocks[idx].getName().endsWith(".meta")) {
+          //
+          // shorten .meta file
+          //
+          RandomAccessFile file = new RandomAccessFile(blocks[idx], "rw");
+          FileChannel channel = file.getChannel();
+          long position = channel.size() - 2;
+          int length = 2;
+          byte[] buffer = new byte[length];
+          random.nextBytes(buffer);
+          channel.write(ByteBuffer.wrap(buffer), position);
+          file.close();
+          LOG.info("Deliberately corrupting file " + blocks[idx].getName() +
+              " at offset " + position + " length " + length);
+
+          // read all files to trigger detection of corrupted replica
+          try {
+            util.checkFiles(fs, "/srcdat10");
+          } catch (BlockMissingException e) {
+            System.out.println("Received BlockMissingException as expected.");
+          } catch (IOException e) {
+            assertTrue("Corrupted replicas not handled properly. Expecting BlockMissingException " +
+                " but received IOException " + e, false);
+          }
+          break;
+        }
+      }
+
+      // fetch bad file list from namenode. There should be one file.
+      badFiles = namenode.getCorruptFiles();
+      LOG.info("Namenode has bad files. " + badFiles.length);
+      assertTrue("Namenode has " + badFiles.length + " bad files. Expecting 1.",
+          badFiles.length == 1);
+      util.cleanup(fs, "/srcdat10");
+    } finally {
+      if (cluster != null) { cluster.shutdown(); }
+    }
+  }
 }