HADOOP-2703. The default options to fsck skips checking files

that are being written to. (lohit vijayarenu via dhruba)



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

CHANGES.txt

@@ -40,6 +40,9 @@ Trunk (unreleased changes)
     HADOOP-3176.  Change lease record when a open-for-write-file 
     gets renamed. (dhruba)
 
+    HADOOP-2703.  The default options to fsck skips checking files
+    that are being written to. (lohit vijayarenu via dhruba) 
+
 Release 0.17.0 - Unreleased
 
   INCOMPATIBLE CHANGES

src/docs/src/documentation/content/xdocs/hdfs_user_guide.xml

@@ -298,7 +298,8 @@
       files, for e.g. missing blocks for a file or under replicated
       blocks. Unlike a traditional fsck utility for native filesystems,
       this command does not correct the errors it detects. Normally Namenode
-      automatically corrects most of the recoverable failures.
+      automatically corrects most of the recoverable failures. By default
+      fsck ignores open files but provides an option to select during reporting.
       HDFS' fsck is not a
       Hadoop shell command. It can be run as '<code>bin/hadoop fsck</code>'.
       Fsck can be run on the whole filesystem or on a subset of files.

src/java/org/apache/hadoop/dfs/ClientProtocol.java

@@ -37,9 +37,9 @@ 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).
-   * 29 : getFileInfo returns null instead of throwing FileNotFoundException
+   * 30 : add underConstruction flag to LocatedBlocks
    */
-  public static final long versionID = 29L;
+  public static final long versionID = 30L;
   
   ///////////////////////////////////////
   // File contents

src/java/org/apache/hadoop/dfs/DFSck.java

@@ -51,6 +51,7 @@ import org.apache.hadoop.util.ToolRunner;
  *  Additionally, the tool collects a detailed overall DFS statistics, and
  *  optionally can print detailed statistics on block locations and replication
  *  factors of each file.
+ *  The tool also provides and option to filter open files during the scan.
  *  
  */
 public class DFSck extends Configured implements Tool {
@@ -77,14 +78,19 @@ public class DFSck extends Configured implements Tool {
   public int run(String[] args) throws Exception {
     String fsName = getInfoServer();
     if (args.length == 0) {
-      System.err.println("Usage: DFSck <path> [-move | -delete] [-files [-blocks [-locations | -racks]]]");
+      System.err.println("Usage: DFSck <path> [-move | -delete | -openforwrite] [-files [-blocks [-locations | -racks]]]");
       System.err.println("\t<path>\tstart checking from this path");
       System.err.println("\t-move\tmove corrupted files to /lost+found");
       System.err.println("\t-delete\tdelete corrupted files");
       System.err.println("\t-files\tprint out files being checked");
+      System.err.println("\t-openforwrite\tprint out files opened for write");
       System.err.println("\t-blocks\tprint out block report");
       System.err.println("\t-locations\tprint out locations for every block");
       System.err.println("\t-racks\tprint out network topology for data-node locations");
+      System.err.println("\t\tBy default fsck ignores files opened for write, " +
+                         "use -openforwrite to report such files. They are usually " +
+                         " tagged CORRUPT or HEALTHY depending on their block " +
+                          "allocation status");
       ToolRunner.printGenericCommandUsage(System.err);
       return -1;
     }
@@ -99,6 +105,7 @@ public class DFSck extends Configured implements Tool {
       if (args[idx].equals("-move")) { url.append("&move=1"); }
       else if (args[idx].equals("-delete")) { url.append("&delete=1"); }
       else if (args[idx].equals("-files")) { url.append("&files=1"); }
+      else if (args[idx].equals("-openforwrite")) { url.append("&openforwrite=1"); }
       else if (args[idx].equals("-blocks")) { url.append("&blocks=1"); }
       else if (args[idx].equals("-locations")) { url.append("&locations=1"); }
       else if (args[idx].equals("-racks")) { url.append("&racks=1"); }

src/java/org/apache/hadoop/dfs/LocatedBlocks.java

@@ -35,15 +35,18 @@ import org.apache.hadoop.io.WritableFactory;
 public class LocatedBlocks implements Writable {
   private long fileLength;
   private List<LocatedBlock> blocks; // array of blocks with prioritized locations
+  private boolean underConstruction;
 
   LocatedBlocks() {
     fileLength = 0;
     blocks = null;
+    underConstruction = false;
   }
   
   LocatedBlocks(INodeFile inode, List<LocatedBlock> blks) {
     fileLength = inode.computeContentSummary().getLength();
     blocks = blks;
+    underConstruction = inode.isUnderConstruction();
   }
   
   /**
@@ -73,6 +76,14 @@ public class LocatedBlocks implements Writable {
   public long getFileLength() {
     return this.fileLength;
   }
+
+  /**
+   * Return ture if file was under construction when 
+   * this LocatedBlocks was constructed, false otherwise.
+   */
+  public boolean isUnderConstruction() {
+    return underConstruction;
+  }
   
   /**
    * Find block containing specified offset.
@@ -148,6 +159,7 @@ public class LocatedBlocks implements Writable {
 
   public void write(DataOutput out) throws IOException {
     out.writeLong(this.fileLength);
+    out.writeBoolean(underConstruction);
     // write located blocks
     int nrBlocks = locatedBlockCount();
     out.writeInt(nrBlocks);
@@ -161,6 +173,7 @@ public class LocatedBlocks implements Writable {
   
   public void readFields(DataInput in) throws IOException {
     this.fileLength = in.readLong();
+    underConstruction = in.readBoolean();
     // read located blocks
     int nrBlocks = in.readInt();
     this.blocks = new ArrayList<LocatedBlock>(nrBlocks);

src/java/org/apache/hadoop/dfs/NamenodeFsck.java

@@ -73,6 +73,7 @@ public class NamenodeFsck {
   private boolean lfInited = false;
   private boolean lfInitedOk = false;
   private boolean showFiles = false;
+  private boolean showOpenFiles = false;
   private boolean showBlocks = false;
   private boolean showLocations = false;
   private boolean showRacks = false;
@@ -106,6 +107,7 @@ public class NamenodeFsck {
       else if (key.equals("blocks")) { this.showBlocks = true; }
       else if (key.equals("locations")) { this.showLocations = true; }
       else if (key.equals("racks")) { this.showRacks = true; }
+      else if (key.equals("openforwrite")) {this.showOpenFiles = true; }
     }
   }
   
@@ -142,6 +144,7 @@ public class NamenodeFsck {
   private void check(DFSFileInfo file, FsckResult res) throws IOException {
     int minReplication = nn.namesystem.getMinReplication();
     String path = file.getPath().toString();
+    boolean isOpen = false;
 
     if (file.isDir()) {
       DFSFileInfo[] files = nn.namesystem.dir.getListing(path);
@@ -157,15 +160,26 @@ public class NamenodeFsck {
       }
       return;
     }
-    res.totalFiles++;
     long fileLen = file.getLen();
-    res.totalSize += fileLen;
     LocatedBlocks blocks = nn.namesystem.getBlockLocations(path, 0, fileLen);
+    isOpen = blocks.isUnderConstruction();
+    if (isOpen && !showOpenFiles) {
+      // We collect these stats about open files to report with default options
+      res.totalOpenFilesSize += fileLen;
+      res.totalOpenFilesBlocks += blocks.locatedBlockCount();
+      res.totalOpenFiles++;
+      return;
+    }
+    res.totalFiles++;
+    res.totalSize += fileLen;
     res.totalBlocks += blocks.locatedBlockCount();
-    if (showFiles) {
+    if (showOpenFiles && isOpen) {
+      out.print(path + " " + fileLen + " bytes, " +
+        blocks.locatedBlockCount() + " block(s), OPENFORWRITE: ");
+    } else if (showFiles) {
       out.print(path + " " + fileLen + " bytes, " +
-          blocks.locatedBlockCount() + " block(s): ");
-    }  else {
+        blocks.locatedBlockCount() + " block(s): ");
+    } else {
       out.print('.');
       out.flush();
       if (res.totalFiles % 100 == 0) { out.println(); }
@@ -250,10 +264,12 @@ public class NamenodeFsck {
       case FIXING_NONE:
         break;
       case FIXING_MOVE:
-        lostFoundMove(file, blocks);
+        if (!isOpen)
+          lostFoundMove(file, blocks);
         break;
       case FIXING_DELETE:
-        nn.namesystem.deleteInternal(path, true, false);
+        if (!isOpen)
+          nn.namesystem.deleteInternal(path, true, false);
       }
     }
     if (showFiles) {
@@ -484,9 +500,12 @@ public class NamenodeFsck {
     private long numMinReplicatedBlocks = 0L;  // minimally replicatedblocks
     private int replication = 0;
     private long totalBlocks = 0L;
+    private long totalOpenFilesBlocks = 0L;
     private long totalFiles = 0L;
+    private long totalOpenFiles = 0L;
     private long totalDirs = 0L;
     private long totalSize = 0L;
+    private long totalOpenFilesSize = 0L;
     private long totalReplicas = 0L;
     private int totalDatanodes = 0;
     private int totalRacks = 0;
@@ -513,7 +532,7 @@ public class NamenodeFsck {
     public long getMissingSize() {
       return missingSize;
     }
-    
+
     public void setMissingSize(long missingSize) {
       this.missingSize = missingSize;
     }
@@ -561,6 +580,16 @@ public class NamenodeFsck {
       this.totalFiles = totalFiles;
     }
     
+    /** Return total number of files opened for write encountered during this scan. */
+    public long getTotalOpenFiles() {
+      return totalOpenFiles;
+    }
+
+    /** Set total number of open files encountered during this scan. */
+    public void setTotalOpenFiles(long totalOpenFiles) {
+      this.totalOpenFiles = totalOpenFiles;
+    }
+    
     /** Return total size of scanned data, in bytes. */
     public long getTotalSize() {
       return totalSize;
@@ -570,6 +599,15 @@ public class NamenodeFsck {
       this.totalSize = totalSize;
     }
     
+    /** Return total size of open files data, in bytes. */
+    public long getTotalOpenFilesSize() {
+      return totalOpenFilesSize;
+    }
+    
+    public void setTotalOpenFilesSize(long totalOpenFilesSize) {
+      this.totalOpenFilesSize = totalOpenFilesSize;
+    }
+    
     /** Return the intended replication factor, against which the over/under-
      * replicated blocks are counted. Note: this values comes from the current
      * Configuration supplied for the tool, so it may be different from the
@@ -592,15 +630,32 @@ public class NamenodeFsck {
       this.totalBlocks = totalBlocks;
     }
     
+    /** Return the total number of blocks held by open files. */
+    public long getTotalOpenFilesBlocks() {
+      return totalOpenFilesBlocks;
+    }
+    
+    public void setTotalOpenFilesBlocks(long totalOpenFilesBlocks) {
+      this.totalOpenFilesBlocks = totalOpenFilesBlocks;
+    }
+    
     public String toString() {
       StringBuffer res = new StringBuffer();
       res.append("Status: " + (isHealthy() ? "HEALTHY" : "CORRUPT"));
       res.append("\n Total size:\t" + totalSize + " B");
+      if (totalOpenFilesSize != 0) 
+        res.append(" (Total open files size: " + totalOpenFilesSize + " B)");
       res.append("\n Total dirs:\t" + totalDirs);
       res.append("\n Total files:\t" + totalFiles);
-      res.append("\n Total blocks:\t" + totalBlocks);
+      if (totalOpenFiles != 0)
+        res.append(" (Files currently being written: " + 
+                   totalOpenFiles + ")");
+      res.append("\n Total blocks (validated):\t" + totalBlocks);
       if (totalBlocks > 0) res.append(" (avg. block size "
                                       + (totalSize / totalBlocks) + " B)");
+      if (totalOpenFilesBlocks != 0)
+        res.append(" (Total open file blocks (not validated): " + 
+                   totalOpenFilesBlocks + ")");
       if (missingSize > 0) {
         res.append("\n  ********************************");
         res.append("\n  CORRUPT FILES:\t" + corruptFiles);

src/test/org/apache/hadoop/dfs/TestFsck.java

@@ -29,6 +29,8 @@ import org.apache.commons.logging.impl.Log4JLogger;
 import org.apache.log4j.Level;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.util.ToolRunner;
 
 /**
@@ -156,4 +158,52 @@ public class TestFsck extends TestCase {
       if (cluster != null) { cluster.shutdown(); }
     }
   }
+  
+  public void testFsckOpenFiles() throws Exception {
+    DFSTestUtil util = new DFSTestUtil("TestFsck", 4, 3, 8*1024); 
+    MiniDFSCluster cluster = null;
+    FileSystem fs = null;
+    try {
+      Configuration conf = new Configuration();
+      cluster = new MiniDFSCluster(conf, 4, true, null);
+      String topDir = "/srcdat";
+      String randomString = "HADOOP  ";
+      fs = cluster.getFileSystem();
+      cluster.waitActive();
+      util.createFiles(fs, topDir);
+      String outStr = runFsck(conf, "/");
+      assertTrue(outStr.contains("HEALTHY"));
+      // Open a file for writing and do not close for now
+      Path openFile = new Path(topDir + "/openFile");
+      FSDataOutputStream out = fs.create(openFile);
+      int writeCount = 0;
+      while (writeCount != 100) {
+        out.write(randomString.getBytes());
+        writeCount++;                  
+      }
+      // We expect the filesystem to be HEALTHY and show one open file
+      outStr = runFsck(conf, topDir);
+      System.out.println(outStr);
+      assertTrue(outStr.contains("HEALTHY"));
+      assertFalse(outStr.contains("OPENFORWRITE")); 
+      // Use -openforwrite option to list open files
+      outStr = runFsck(conf, topDir, "-openforwrite");
+      System.out.println(outStr);
+      assertTrue(outStr.contains("OPENFORWRITE"));
+      assertTrue(outStr.contains("openFile"));
+      // Close the file
+      out.close(); 
+      // Now, fsck should show HEALTHY fs and should not show any open files
+      outStr = runFsck(conf, topDir);
+      System.out.println(outStr);
+      assertTrue(outStr.contains("HEALTHY"));
+      assertFalse(outStr.contains("OPENFORWRITE"));
+      util.cleanup(fs, topDir);
+      if (fs != null) {try{fs.close();} catch(Exception e){}}
+      cluster.shutdown();
+    } finally {
+      if (fs != null) {try{fs.close();} catch(Exception e){}}
+      if (cluster != null) { cluster.shutdown(); }
+    }
+  }
 }