HDFS-3274. Add a new conf key dfs.journal.edits.dir and use it for edit log initialization. Contributed by Hari Mankude

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/HDFS-3092@1326010 13f79535-47bb-0310-9956-ffa450edef68

hadoop-hdfs-project/hadoop-hdfs/CHANGES.HDFS-3092.txt

@@ -9,9 +9,12 @@ HDFS-3092 branch changes
     HDFS-3185. Add configuration of Journal Service Daemons. 
     (Hari Mankude via suresh)
 
-    HDFS-3213 Persist the cluster id and nsid in JournalService storage
+    HDFS-3213. Persist the cluster id and nsid in JournalService storage
     directory.  (Hari Mankude via szetszwo)
 
+    HDFS-3274. Add a new conf key dfs.journal.edits.dir and use it for
+    edit log initialization.  (Hari Mankude via szetszwo)
+
   IMPROVEMENTS
 
   OPTIMIZATIONS

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java

@@ -203,6 +203,7 @@ public class DFSConfigKeys extends CommonConfigurationKeys {
   
   // This is a comma separated host:port list of addresses hosting the journal service
   public static final String  DFS_JOURNALNODE_ADDRESS_KEY = "dfs.journalnode.addresses";
+  public static final String  DFS_JOURNAL_EDITS_DIR_KEY = "dfs.journal.edits.dir";
 
   // Much code in hdfs is not yet updated to use these keys.
   public static final String  DFS_CLIENT_BLOCK_WRITE_LOCATEFOLLOWINGBLOCK_RETRIES_KEY = "dfs.client.block.write.locateFollowingBlock.retries";

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/journalservice/JournalService.java

@@ -17,12 +17,19 @@
  */
 package org.apache.hadoop.hdfs.server.journalservice;
 
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_SHARED_EDITS_DIR_KEY;
+
+import java.io.File;
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.net.URI;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.HashMap;
+import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
+import java.util.Properties;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -34,10 +41,15 @@ import org.apache.hadoop.hdfs.protocol.proto.JournalProtocolProtos.JournalProtoc
 import org.apache.hadoop.hdfs.protocolPB.JournalProtocolPB;
 import org.apache.hadoop.hdfs.protocolPB.JournalProtocolServerSideTranslatorPB;
 import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.NamenodeRole;
+import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.StartupOption;
+import org.apache.hadoop.hdfs.server.common.Storage.StorageDirectory;
+import org.apache.hadoop.hdfs.server.common.Storage.StorageState;
+import org.apache.hadoop.hdfs.server.common.Storage;
 import org.apache.hadoop.hdfs.server.common.StorageInfo;
+import org.apache.hadoop.hdfs.server.common.Util;
 import org.apache.hadoop.hdfs.server.namenode.FSImage;
-import org.apache.hadoop.hdfs.server.namenode.FSNamesystem;
 import org.apache.hadoop.hdfs.server.namenode.NNStorage;
+import org.apache.hadoop.hdfs.server.namenode.FSEditLog;
 import org.apache.hadoop.hdfs.server.protocol.FenceResponse;
 import org.apache.hadoop.hdfs.server.protocol.FencedException;
 import org.apache.hadoop.hdfs.server.protocol.JournalInfo;
@@ -53,6 +65,8 @@ import org.apache.hadoop.security.UserGroupInformation;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.protobuf.BlockingService;
 
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_JOURNAL_EDITS_DIR_KEY;
+
 /**
  * This class interfaces with the namenode using {@link JournalProtocol} over
  * RPC. It has two modes: <br>
@@ -81,6 +95,8 @@ public class JournalService implements JournalProtocol {
   private String fencerInfo;
   private StorageInfo storageInfo;
   private Configuration conf;
+  private FSEditLog editLog;
+  private FSImage image;
   
   enum State {
     /** The service is initialized and ready to start. */
@@ -182,16 +198,59 @@ public class JournalService implements JournalProtocol {
         .getProxy();
     this.rpcServer = createRpcServer(conf, serverAddress, this);
     this.conf = conf;
+ 
+    try {
+      initializeJournalStorage(conf);
+    } catch (IOException ioe) {
+      LOG.info("Exception in initialize: " + ioe.getMessage());
+      throw ioe;
+    }
+  }
   
+  /** This routine initializes the storage directory. It is possible that
+   *  the directory is not formatted. In this case, it creates dummy entries
+   *  and storage is later formatted.
+   */
+  private void initializeJournalStorage(Configuration conf) throws IOException {
+    
+    boolean isFormatted = false;
+    Collection<URI> dirsToFormat = new ArrayList<URI>();
+    List<URI> editUrisToFormat = getJournalEditsDirs(conf);
+
     // Load the newly formatted image, using all of the directories
-    FSImage image = new FSImage(conf);
-    storageInfo = image.getStorage();
-    LOG.info("JournalService constructor, nsid " + storageInfo.getNamespaceID()
-        + " cluster id " + storageInfo.getClusterID());
-
-    String addr = NetUtils.getHostPortString(rpcServer.getListenerAddress());
-    registration = new NamenodeRegistration(addr, "", storageInfo,
-        NamenodeRole.BACKUP);
+    image = new FSImage(conf, dirsToFormat, editUrisToFormat);
+    Map<StorageDirectory, StorageState> dataDirStates =
+          new HashMap<StorageDirectory, StorageState>();
+    isFormatted = image
+        .recoverStorageDirs(StartupOption.REGULAR, dataDirStates);
+
+    if (isFormatted == true) {
+      // Directory has been formatted. So, it should have a versionfile.
+      this.editLog = image.getEditLog();
+      Iterator<StorageDirectory> sdit = image.getStorage().dirIterator(
+          NNStorage.NameNodeDirType.IMAGE);
+      StorageDirectory sd = sdit.next();
+
+      Properties props = Storage.readPropertiesFile(sd.getVersionFile());
+      String cid = props.getProperty("clusterID");
+      String nsid = props.getProperty("namespaceID");
+      String layout = props.getProperty("layoutVersion");
+      storageInfo = new StorageInfo(Integer.parseInt(layout),
+          Integer.parseInt(nsid), cid, 0);
+
+      LOG.info("JournalService constructor, nsid "
+          + storageInfo.getNamespaceID() + " cluster id "
+          + storageInfo.getClusterID());
+
+      String addr = NetUtils.getHostPortString(rpcServer.getListenerAddress());
+      registration = new NamenodeRegistration(addr, "", storageInfo,
+          NamenodeRole.BACKUP);
+    } else {
+      // Storage directory has not been formatted. So create dummy entries for now.
+      image = new FSImage(conf);
+      storageInfo = image.getStorage();
+      editLog = null;
+    }
   }
   
   /**
@@ -275,24 +334,27 @@ public class JournalService implements JournalProtocol {
   }
   
   private void setupStorage(JournalInfo jinfo) throws IOException {
-    setupStorage(jinfo.getNamespaceId(), jinfo.getClusterId());
+    formatStorage(jinfo.getNamespaceId(), jinfo.getClusterId());
   }
   
   private void setupStorage(NamespaceInfo nsinfo) throws IOException {
-    setupStorage(nsinfo.getNamespaceID(), nsinfo.getClusterID());
+    formatStorage(nsinfo.getNamespaceID(), nsinfo.getClusterID());
   }
   
-  // Will format the edits dir and save the nsid + cluster id.
-  private void setupStorage(int nsId, String clusterId) throws IOException {
-    // For now, use the namespacedirs and edits dir to save the journal info.
-    // Going forward, this can be modified to extract journal specific edits
-    // dir.
+  /**
+   * Will format the edits dir and save the nsid + cluster id.
+   * @param nsId
+   * @param clusterId
+   * @throws IOException
+   */
+  private void formatStorage(int nsId, String clusterId) throws IOException {
     Collection<URI> dirsToFormat = new ArrayList<URI>();
-    List<URI> editUrisToFormat = FSNamesystem
-        .getNamespaceEditsDirs(conf, false);
+    List<URI> editUrisToFormat = getJournalEditsDirs(conf);
     NNStorage nnStorage = new NNStorage(conf, dirsToFormat, editUrisToFormat);
     LOG.info("Setting up storage for nsid " + nsId + " clusterid " + clusterId);
     nnStorage.format(new NamespaceInfo(nsId, clusterId, "journalservice", 0, 0));
+    image = new FSImage(conf, dirsToFormat, editUrisToFormat); 
+    this.editLog = image.getEditLog();
 
     storageInfo = new StorageInfo(LayoutVersion.getCurrentLayoutVersion(),
         nsId, clusterId, 0);
@@ -422,4 +484,17 @@ public class JournalService implements JournalProtocol {
   long getEpoch() {
     return epoch;
   }
+  
+  /**
+   * Returns edit directories that are shared between primary and secondary.
+   * @param conf
+   * @return Collection of edit directories.
+   */
+  public List<URI> getJournalEditsDirs(Configuration conf) {
+    // don't use getStorageDirs here, because we want an empty default
+    // rather than the dir in /tmp
+    Collection<String> dirNames = conf.getTrimmedStringCollection(
+        DFS_JOURNAL_EDITS_DIR_KEY);
+    return Util.stringCollectionAsURIs(dirNames);
+  }
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSImage.java

@@ -114,7 +114,7 @@ public class FSImage implements Closeable {
    * @param editsDirs Directories the editlog can be stored in.
    * @throws IOException if directories are invalid.
    */
-  protected FSImage(Configuration conf,
+  public FSImage(Configuration conf,
                     Collection<URI> imageDirs,
                     List<URI> editsDirs)
       throws IOException {
@@ -254,7 +254,7 @@ public class FSImage implements Closeable {
    * @param dataDirStates output of storage directory states
    * @return true if there is at least one valid formatted storage directory
    */
-  private boolean recoverStorageDirs(StartupOption startOpt,
+  public boolean recoverStorageDirs(StartupOption startOpt,
       Map<StorageDirectory, StorageState> dataDirStates) throws IOException {
     boolean isFormatted = false;
     for (Iterator<StorageDirectory> it = 

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NNStorage.java

@@ -93,7 +93,7 @@ public class NNStorage extends Storage implements Closeable {
    * or of type EDITS which stores edits or of type IMAGE_AND_EDITS which
    * stores both fsimage and edits.
    */
-  static enum NameNodeDirType implements StorageDirType {
+  public static enum NameNodeDirType implements StorageDirType {
     UNDEFINED,
     IMAGE,
     EDITS,

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/MiniDFSCluster.java

@@ -702,6 +702,15 @@ public class MiniDFSCluster {
         minNN + "-through-" + maxNN));
   }
   
+  public File getJournalEditsDir() throws IOException {
+    return formatJournalEditsDir(base_dir);
+  }
+  
+  public static File formatJournalEditsDir(File baseDir)
+      throws IOException {
+    return (new File(baseDir, "journal-edits"));
+  }
+  
   public NameNodeInfo[] getNameNodeInfos() {
     return this.nameNodes;
   }

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/journalservice/TestJournalService.java

@@ -17,8 +17,10 @@
  */
 package org.apache.hadoop.hdfs.server.journalservice;
 
+import java.io.File;
 import java.io.IOException;
 import java.net.InetSocketAddress;
+import java.net.URI;
 
 import junit.framework.Assert;
 
@@ -32,9 +34,11 @@ import org.apache.hadoop.hdfs.HdfsConfiguration;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.apache.hadoop.hdfs.protocol.UnregisteredNodeException;
 import org.apache.hadoop.hdfs.server.namenode.NameNode;
+import org.apache.hadoop.hdfs.server.namenode.ha.HATestUtil;
 import org.apache.hadoop.hdfs.server.protocol.FenceResponse;
 import org.apache.hadoop.hdfs.server.protocol.FencedException;
 import org.apache.hadoop.hdfs.server.protocol.JournalInfo;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.*;
 import org.junit.Test;
 import org.mockito.Mockito;
 
@@ -58,10 +62,14 @@ public class TestJournalService {
     try {
       cluster = new MiniDFSCluster.Builder(conf).numDataNodes(1).build();
       cluster.waitActive(0);
+      FileSystem fs = FileSystem.getLocal(conf);
+      File journalEditsDir = cluster.getJournalEditsDir();
+      boolean result = fs.mkdirs(new Path(journalEditsDir.toString()));
+      conf.set(DFS_JOURNAL_EDITS_DIR_KEY, journalEditsDir.toString());
       service = startJournalService(listener);
       verifyRollLogsCallback(service, listener);
       verifyJournalCallback(service, listener);
-      verifyFence(service, cluster.getNameNode(0));
+      verifyFence(service, listener, cluster.getNameNode(0));
     } finally {
       if (service != null) {
         stopJournalService(service);
@@ -71,6 +79,12 @@ public class TestJournalService {
       }
     }
   }
+  
+  private JournalService restartJournalService(JournalService service,
+      JournalListener listener) throws IOException {
+    stopJournalService(service);
+    return (startJournalService(listener));
+  }
 
   private JournalService startJournalService(JournalListener listener)
       throws IOException {
@@ -109,7 +123,7 @@ public class TestJournalService {
         Mockito.anyLong(), Mockito.anyInt(), (byte[]) Mockito.any());
   }
   
-  public void verifyFence(JournalService s, NameNode nn) throws Exception {
+  public void verifyFence(JournalService s, JournalListener listener, NameNode nn) throws Exception {
     String cid = nn.getNamesystem().getClusterId();
     int nsId = nn.getNamesystem().getFSImage().getNamespaceID();
     int lv = nn.getNamesystem().getFSImage().getLayoutVersion();
@@ -156,6 +170,11 @@ public class TestJournalService {
     } catch (UnregisteredNodeException ignore) {
       LOG.info(ignore.getMessage());
     } 
+    
+    s = restartJournalService(s, listener);
+    // New epoch higher than the current epoch is successful
+    resp = s.fence(info, currentEpoch+1, "fencer");
+    Assert.assertNotNull(resp);
    
   }
 }