HADOOP-4885. Try to restore failed name-node storage directories at checkpoint time. Contributed by Boris Shkolnik.

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

CHANGES.txt

@@ -104,6 +104,9 @@ Trunk (unreleased changes)
     HADOOP-4220. Changes the JobTracker restart tests so that they take much
     less time. (Amar Kamat via ddas)
 
+    HADOOP-4885. Try to restore failed name-node storage directories at 
+    checkpoint time. (Boris Shkolnik via shv)
+
   OPTIMIZATIONS
 
   BUG FIXES

src/hdfs/org/apache/hadoop/hdfs/server/common/Storage.java

@@ -174,6 +174,17 @@ public abstract class Storage extends StorageInfo {
     return new DirIterator(dirType);
   }
   
+  /**
+   * generate storage list (debug line)
+   */
+  public String listStorageDirectories() {
+    StringBuffer buf = new StringBuffer();
+    for (StorageDirectory sd : storageDirs) {
+      buf.append(sd.getRoot() + "(" + sd.getStorageDirType() + ");");
+    }
+    return buf.toString();
+  }
+  
   /**
    * One of the storage directories.
    */

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

@@ -55,6 +55,10 @@ class FSDirectory implements Closeable {
   /** Access an existing dfs name directory. */
   FSDirectory(FSNamesystem ns, Configuration conf) {
     this(new FSImage(), ns, conf);
+    if(conf.getBoolean("dfs.name.dir.restore", false)) {
+      NameNode.LOG.info("set FSImage.restoreFailedStorage");
+      fsImage.setRestoreFailedStorage(true);
+    }
     fsImage.setCheckpointDirectories(FSImage.getCheckpointDirs(conf, null),
                                 FSImage.getCheckpointEditsDirs(conf, null));
   }

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

@@ -332,6 +332,7 @@ public class FSEditLog {
       } catch (IOException e) {
         FSNamesystem.LOG.warn("Unable to open edit log file " + eFile);
         // Remove the directory from list of storage directories
+        fsimage.removedStorageDirs.add(sd);
         it.remove();
       }
     }
@@ -378,6 +379,8 @@ public class FSEditLog {
         eStream.flush();
         eStream.close();
       } catch (IOException e) {
+        FSNamesystem.LOG.warn("FSEditLog:close - failed to close stream " 
+            + eStream.getName());
         processIOError(idx);
         idx--;
       }
@@ -399,9 +402,15 @@ public class FSEditLog {
     assert(index < getNumStorageDirs());
     assert(getNumStorageDirs() == editStreams.size());
     
+    EditLogFileOutputStream eStream = (EditLogFileOutputStream)editStreams.get(index);
     File parentStorageDir = ((EditLogFileOutputStream)editStreams
                                       .get(index)).getFile()
                                       .getParentFile().getParentFile();
+    
+    try {
+      eStream.close();
+    } catch (Exception e) {}
+    
     editStreams.remove(index);
     //
     // Invoke the ioerror routine of the fsimage
@@ -840,6 +849,7 @@ public class FSEditLog {
       try {
         eStream.write(op, writables);
       } catch (IOException ie) {
+        FSImage.LOG.warn("logEdit: removing "+ eStream.getName(), ie);
         processIOError(idx);         
         // processIOError will remove the idx's stream 
         // from the editStreams collection, so we need to update idx
@@ -1105,9 +1115,16 @@ public class FSEditLog {
     assert(getNumStorageDirs() == editStreams.size());
     long size = 0;
     for (int idx = 0; idx < editStreams.size(); idx++) {
-      long curSize = editStreams.get(idx).length();
-      assert (size == 0 || size == curSize) : "All streams must be the same";
-      size = curSize;
+      EditLogOutputStream es = editStreams.get(idx);
+      try {
+        long curSize = es.length();
+        assert (size == 0 || size == curSize) : "All streams must be the same";
+        size = curSize;
+      } catch (IOException e) {
+        FSImage.LOG.warn("getEditLogSize: editstream.length failed. removing editlog (" +
+            idx + ") " + es.getName());
+        processIOError(idx);
+      }
     }
     return size;
   }
@@ -1135,9 +1152,13 @@ public class FSEditLog {
 
     close();                     // close existing edit log
 
+    // check if any of failed storage is now available and put it back
+    fsimage.attemptRestoreRemovedStorage();
+    
     //
     // Open edits.new
     //
+    boolean failedSd = false;
     for (Iterator<StorageDirectory> it = 
            fsimage.dirIterator(NameNodeDirType.EDITS); it.hasNext();) {
       StorageDirectory sd = it.next();
@@ -1147,11 +1168,16 @@ public class FSEditLog {
         eStream.create();
         editStreams.add(eStream);
       } catch (IOException e) {
+        failedSd = true;
         // remove stream and this storage directory from list
-        processIOError(sd);
-       it.remove();
+        FSImage.LOG.warn("rollEdidLog: removing storage " + sd.getRoot().getPath());
+        sd.unlock();
+        fsimage.removedStorageDirs.add(sd);
+        it.remove();
       }
     }
+    if(failedSd)
+      fsimage.incrementCheckpointTime();  // update time for the valid ones
   }
 
   /**
@@ -1182,6 +1208,8 @@ public class FSEditLog {
         getEditFile(sd).delete();
         if (!getEditNewFile(sd).renameTo(getEditFile(sd))) {
           // Should we also remove from edits
+          NameNode.LOG.warn("purgeEditLog: removing failed storage " + sd.getRoot().getPath());
+          fsimage.removedStorageDirs.add(sd);
           it.remove(); 
         }
       }

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

@@ -41,6 +41,7 @@ import java.util.HashMap;
 import java.lang.Math;
 import java.nio.ByteBuffer;
 
+import org.apache.hadoop.fs.FileUtil;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.permission.PermissionStatus;
 import org.apache.hadoop.fs.permission.FsPermission;
@@ -112,6 +113,19 @@ public class FSImage extends Storage {
   protected long checkpointTime = -1L;
   private FSEditLog editLog = null;
   private boolean isUpgradeFinalized = false;
+
+  /**
+   * flag that controls if we try to restore failed storages
+   */
+  private boolean restoreFailedStorage = false;
+  public void setRestoreFailedStorage(boolean val) {
+    LOG.info("enabled failed storage replicas restore");
+    restoreFailedStorage=val;
+  }
+  
+  public boolean getRestoreFailedStorage() {
+    return restoreFailedStorage;
+  }
   
   /**
    * list of failed (and thus removed) storages
@@ -624,12 +638,13 @@ public class FSImage extends Storage {
         writeCheckpointTime(sd);
       } catch(IOException e) {
         // Close any edits stream associated with this dir and remove directory
-     if (sd.getStorageDirType().isOfType(NameNodeDirType.EDITS))
-       editLog.processIOError(sd);
-     
-   //add storage to the removed list
-     removedStorageDirs.add(sd);
-     it.remove();
+        LOG.warn("incrementCheckpointTime failed on " + sd.getRoot().getPath() + ";type="+sd.getStorageDirType());
+        if (sd.getStorageDirType().isOfType(NameNodeDirType.EDITS))
+          editLog.processIOError(sd);
+
+        //add storage to the removed list
+        removedStorageDirs.add(sd);
+        it.remove();
       }
     }
   }
@@ -644,7 +659,10 @@ public class FSImage extends Storage {
       StorageDirectory sd = it.next();
       if (sd.getRoot().getPath().equals(dirName.getPath())) {
         //add storage to the removed list
-        LOG.info(" removing " + dirName.getPath());
+        LOG.warn("FSImage:processIOError: removing storage: " + dirName.getPath());
+        try {
+          sd.unlock(); //try to unlock before removing (in case it is restored)
+        } catch (Exception e) {}
         removedStorageDirs.add(sd);
         it.remove();
       }
@@ -1304,7 +1322,7 @@ public class FSImage extends Storage {
       }
     }
     editLog.purgeEditLog(); // renamed edits.new to edits
-
+    LOG.debug("rollFSImage after purgeEditLog: storageList=" + listStorageDirectories());
     //
     // Renames new image
     //
@@ -1315,13 +1333,18 @@ public class FSImage extends Storage {
       File curFile = getImageFile(sd, NameNodeFile.IMAGE);
       // renameTo fails on Windows if the destination file 
       // already exists.
+      LOG.debug("renaming  " + ckpt.getAbsolutePath() + " to "  + curFile.getAbsolutePath());
       if (!ckpt.renameTo(curFile)) {
         curFile.delete();
         if (!ckpt.renameTo(curFile)) {
+          LOG.warn("renaming  " + ckpt.getAbsolutePath() + " to "  + 
+              curFile.getAbsolutePath() + " FAILED");
+          
           // Close edit stream, if this directory is also used for edits
           if (sd.getStorageDirType().isOfType(NameNodeDirType.EDITS))
             editLog.processIOError(sd);
-        // add storage to the removed list
+          
+          // add storage to the removed list
           removedStorageDirs.add(sd);
           it.remove();
         }
@@ -1414,6 +1437,36 @@ public class FSImage extends Storage {
   return getImageFile(sd, NameNodeFile.IMAGE); 
   }
 
+  /**
+   * See if any of removed storages iw "writable" again, and can be returned 
+   * into service
+   */
+  void attemptRestoreRemovedStorage() {   
+    // if directory is "alive" - copy the images there...
+    if(!restoreFailedStorage || removedStorageDirs.size() == 0) 
+      return; //nothing to restore
+    
+    LOG.info("FSImage.attemptRestoreRemovedStorage: check removed(failed) " +
+    		"storarge. removedStorages size = " + removedStorageDirs.size());
+    for(Iterator<StorageDirectory> it = this.removedStorageDirs.iterator(); it.hasNext();) {
+      StorageDirectory sd = it.next();
+      File root = sd.getRoot();
+      LOG.info("currently disabled dir " + root.getAbsolutePath() + 
+          "; type="+sd.getStorageDirType() + ";canwrite="+root.canWrite());
+      try {
+        
+        if(root.exists() && root.canWrite()) { 
+          format(sd);
+          LOG.info("restoring dir " + sd.getRoot().getAbsolutePath());
+          this.addStorageDir(sd); // restore
+          it.remove();
+        }
+      } catch(IOException e) {
+        LOG.warn("failed to restore " + sd.getRoot().getAbsolutePath(), e);
+      }
+    }    
+  }
+  
   public File getFsEditName() throws IOException {
     return getEditLog().getFsEditName();
   }

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

@@ -318,7 +318,7 @@ public class SecondaryNameNode implements Runnable {
     startCheckpoint();
 
     // Tell the namenode to start logging transactions in a new edit file
-    // Retuns a token that would be used to upload the merged image.
+    // Returns a token that would be used to upload the merged image.
     CheckpointSignature sig = (CheckpointSignature)namenode.rollEditLog();
 
     // error simulation code for junit test