HDFS-16690. Automatically format unformatted JNs with JournalNodeSyncer

aswinmprabhu · Aswin M Prabhu · commit 318a28121a52 · 2024-07-04T16:07:32.000+05:30
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/protocol/InterQJournalProtocol.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/protocol/InterQJournalProtocol.java
@@ -20,6 +20,7 @@
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.hdfs.DFSConfigKeys;
+import org.apache.hadoop.hdfs.protocol.proto.HdfsServerProtos.StorageInfoProto;
 import org.apache.hadoop.hdfs.qjournal.server.JournalNode;
 import org.apache.hadoop.hdfs.qjournal.protocol.QJournalProtocolProtos.GetEditLogManifestResponseProto;
 import org.apache.hadoop.security.KerberosInfo;
@@ -51,4 +52,13 @@ GetEditLogManifestResponseProto getEditLogManifestFromJournal(
       String jid, String nameServiceId, long sinceTxId, boolean inProgressOk)
       throws IOException;
 
+  /**
+   * Get the storage info for the specified journal.
+   * @param jid the journal identifier
+   * @param nameServiceId the name service id
+   * @return the storage info object
+   */
+  StorageInfoProto getStorageInfo(String jid, String nameServiceId)
+      throws IOException;
+
 }
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/protocolPB/InterQJournalProtocolServerSideTranslatorPB.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/protocolPB/InterQJournalProtocolServerSideTranslatorPB.java
@@ -24,6 +24,8 @@
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.hdfs.qjournal.protocol.InterQJournalProtocol;
+import org.apache.hadoop.hdfs.protocol.proto.HdfsServerProtos.StorageInfoProto;
+import org.apache.hadoop.hdfs.qjournal.protocol.InterQJournalProtocolProtos.GetStorageInfoRequestProto;
 import org.apache.hadoop.hdfs.qjournal.protocol.QJournalProtocolProtos.GetEditLogManifestRequestProto;
 import org.apache.hadoop.hdfs.qjournal.protocol.QJournalProtocolProtos.GetEditLogManifestResponseProto;
 
@@ -60,4 +62,18 @@ public GetEditLogManifestResponseProto getEditLogManifestFromJournal(
       throw new ServiceException(e);
     }
   }
+
+  @Override
+  public StorageInfoProto getStorageInfo(
+      RpcController controller, GetStorageInfoRequestProto request)
+      throws ServiceException {
+    try {
+      return impl.getStorageInfo(
+          request.getJid().getIdentifier(),
+          request.hasNameServiceId() ? request.getNameServiceId() : null
+      );
+    } catch (IOException e) {
+      throw new ServiceException(e);
+    }
+  }
 }
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/protocolPB/InterQJournalProtocolTranslatorPB.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/protocolPB/InterQJournalProtocolTranslatorPB.java
@@ -19,6 +19,8 @@
 
 package org.apache.hadoop.hdfs.qjournal.protocolPB;
 
+import org.apache.hadoop.hdfs.protocol.proto.HdfsServerProtos.StorageInfoProto;
+import org.apache.hadoop.hdfs.qjournal.protocol.InterQJournalProtocolProtos;
 import org.apache.hadoop.thirdparty.protobuf.RpcController;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
@@ -75,6 +77,18 @@ public GetEditLogManifestResponseProto getEditLogManifestFromJournal(
         req.build()));
   }
 
+  @Override
+  public StorageInfoProto getStorageInfo(String jid, String nameServiceId)
+      throws IOException {
+    InterQJournalProtocolProtos.GetStorageInfoRequestProto.Builder req =
+        InterQJournalProtocolProtos.GetStorageInfoRequestProto.newBuilder()
+            .setJid(convertJournalId(jid));
+    if (nameServiceId != null) {
+      req.setNameServiceId(nameServiceId);
+    }
+    return ipc(() -> rpcProxy.getStorageInfo(NULL_CONTROLLER, req.build()));
+  }
+
   private QJournalProtocolProtos.JournalIdProto convertJournalId(String jid) {
     return QJournalProtocolProtos.JournalIdProto.newBuilder()
         .setIdentifier(jid)
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/server/JournalNodeRpcServer.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/server/JournalNodeRpcServer.java
@@ -18,6 +18,7 @@
 package org.apache.hadoop.hdfs.qjournal.server;
 
 import org.apache.hadoop.classification.VisibleForTesting;
+import org.apache.hadoop.hdfs.protocol.proto.HdfsServerProtos.StorageInfoProto;
 import org.apache.hadoop.thirdparty.protobuf.BlockingService;
 import org.slf4j.Logger;
 import org.apache.hadoop.classification.InterfaceAudience;
@@ -71,14 +72,14 @@ public class JournalNodeRpcServer implements QJournalProtocol,
 
   JournalNodeRpcServer(Configuration conf, JournalNode jn) throws IOException {
     this.jn = jn;
-    
+
     Configuration confCopy = new Configuration(conf);
-    
+
     // Ensure that nagling doesn't kick in, which could cause latency issues.
     confCopy.setBoolean(
         CommonConfigurationKeysPublic.IPC_SERVER_TCPNODELAY_KEY,
         true);
-    
+
     InetSocketAddress addr = getAddress(confCopy);
     String bindHost = conf.getTrimmed(DFS_JOURNALNODE_RPC_BIND_HOST_KEY, null);
     if (bindHost == null) {
@@ -104,7 +105,7 @@ public class JournalNodeRpcServer implements QJournalProtocol,
     this.handlerCount = confHandlerCount;
     LOG.info("The number of JournalNodeRpcServer handlers is {}.",
         this.handlerCount);
-    
+
     this.server = new RPC.Builder(confCopy)
         .setProtocol(QJournalProtocolPB.class)
         .setInstance(service)
@@ -149,15 +150,15 @@ void start() {
   public InetSocketAddress getAddress() {
     return server.getListenerAddress();
   }
-  
+
   void join() throws InterruptedException {
     this.server.join();
   }
-  
+
   void stop() {
     this.server.stop();
   }
-  
+
   static InetSocketAddress getAddress(Configuration conf) {
     String addr = conf.get(
         DFSConfigKeys.DFS_JOURNALNODE_RPC_ADDRESS_KEY,
@@ -211,7 +212,7 @@ public void journal(RequestInfo reqInfo,
     jn.getOrCreateJournal(reqInfo.getJournalId(), reqInfo.getNameServiceId())
        .journal(reqInfo, segmentTxId, firstTxnId, numTxns, records);
   }
-  
+
   @Override
   public void heartbeat(RequestInfo reqInfo) throws IOException {
     jn.getOrCreateJournal(reqInfo.getJournalId(), reqInfo.getNameServiceId())
@@ -245,17 +246,24 @@ public GetEditLogManifestResponseProto getEditLogManifest(
       String jid, String nameServiceId,
       long sinceTxId, boolean inProgressOk)
       throws IOException {
-    
+
     RemoteEditLogManifest manifest = jn.getOrCreateJournal(jid, nameServiceId)
         .getEditLogManifest(sinceTxId, inProgressOk);
-    
+
     return GetEditLogManifestResponseProto.newBuilder()
         .setManifest(PBHelper.convert(manifest))
         .setHttpPort(jn.getBoundHttpAddress().getPort())
         .setFromURL(jn.getHttpServerURI())
         .build();
   }
 
+  @Override
+  public StorageInfoProto getStorageInfo(String jid,
+      String nameServiceId) throws IOException {
+    StorageInfo storage = jn.getOrCreateJournal(jid).getStorage();
+    return PBHelper.convert(storage);
+  }
+
   @Override
   public GetJournaledEditsResponseProto getJournaledEdits(String jid,
       String nameServiceId, long sinceTxId, int maxTxns) throws IOException {
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/server/JournalNodeSyncer.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/server/JournalNodeSyncer.java
@@ -18,6 +18,9 @@
 package org.apache.hadoop.hdfs.qjournal.server;
 
 import org.apache.hadoop.classification.VisibleForTesting;
+import org.apache.hadoop.hdfs.protocol.proto.HdfsServerProtos;
+import org.apache.hadoop.hdfs.server.common.HdfsServerConstants;
+import org.apache.hadoop.hdfs.server.common.StorageInfo;
 import org.apache.hadoop.thirdparty.com.google.common.collect.ImmutableList;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.conf.Configuration;
@@ -168,6 +171,10 @@ private boolean getOtherJournalNodeProxies() {
 
   private void startSyncJournalsDaemon() {
     syncJournalDaemon = new Daemon(() -> {
+      // Format the journal with namespace info from the other JNs if it is not formatted
+      if (!journal.isFormatted()) {
+        formatWithSyncer();
+      }
       // Wait for journal to be formatted to create edits.sync directory
       while(!journal.isFormatted()) {
         try {
@@ -187,7 +194,8 @@ private void startSyncJournalsDaemon() {
       while(shouldSync) {
         try {
           if (!journal.isFormatted()) {
-            LOG.warn("Journal cannot sync. Not formatted.");
+            LOG.warn("Journal cannot sync. Not formatted. Trying to format with the syncer");
+            formatWithSyncer();
           } else {
             syncJournals();
           }
@@ -233,6 +241,36 @@ private void syncJournals() {
     journalNodeIndexForSync = (journalNodeIndexForSync + 1) % numOtherJNs;
   }
 
+  private void formatWithSyncer() {
+    LOG.info("Trying to format the journal with the syncer");
+    try {
+      StorageInfo storage = null;
+      for (JournalNodeProxy jnProxy : otherJNProxies) {
+        try {
+          HdfsServerProtos.StorageInfoProto storageInfoResponse = jnProxy.jnProxy.getStorageInfo(jid, null);
+          storage = PBHelper.convert(storageInfoResponse, HdfsServerConstants.NodeType.JOURNAL_NODE);
+          if (storage.getNamespaceID() == 0) {
+            LOG.error("Got invalid StorageInfo from " + jnProxy);
+            continue;
+          }
+          LOG.info("Got StorageInfo " + storage + " from " + jnProxy);
+          break;
+        } catch (IOException e) {
+          LOG.error("Could not get StorageInfo from " + jnProxy, e);
+        }
+      }
+      if (storage == null) {
+        LOG.error("Could not get StorageInfo from any JournalNode. " +
+            "JournalNodeSyncer cannot format the journal.");
+        return;
+      }
+      NamespaceInfo nsInfo = new NamespaceInfo(storage);
+      journal.format(nsInfo, true);
+    } catch (IOException e) {
+      LOG.error("Exception in formatting the journal with the syncer", e);
+    }
+  }
+
   private void syncWithJournalAtIndex(int index) {
     LOG.info("Syncing Journal " + jn.getBoundIpcAddress().getAddress() + ":"
         + jn.getBoundIpcAddress().getPort() + " with "
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/proto/InterQJournalProtocol.proto b/hadoop-hdfs-project/hadoop-hdfs/src/main/proto/InterQJournalProtocol.proto
@@ -31,8 +31,15 @@ package hadoop.hdfs.qjournal;
 import "HdfsServer.proto";
 import "QJournalProtocol.proto";
 
+message GetStorageInfoRequestProto {
+    required JournalIdProto jid = 1;
+    optional string nameServiceId = 2;
+}
 
 service InterQJournalProtocolService {
     rpc getEditLogManifestFromJournal(GetEditLogManifestRequestProto)
     returns (GetEditLogManifestResponseProto);
+
+    rpc getStorageInfo(GetStorageInfoRequestProto)
+    returns (StorageInfoProto);
 }
