HDDS-1371. OMSnapshotProvider to download DB checkpoint from leader OM. (#703)

Author: hanishakoneru

hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java

@@ -76,6 +76,12 @@ public final class OzoneConsts {
   public static final String OZONE_USER = "user";
   public static final String OZONE_REQUEST = "request";
 
+  // OM Http server endpoints
+  public static final String OZONE_OM_SERVICE_LIST_HTTP_ENDPOINT =
+      "/serviceList";
+  public static final String OZONE_OM_DB_CHECKPOINT_HTTP_ENDPOINT =
+      "/dbCheckpoint";
+
   // Ozone File System scheme
   public static final String OZONE_URI_SCHEME = "o3fs";
 
@@ -286,4 +292,9 @@ private OzoneConsts() {
 
   // OM Ratis snapshot file to store the last applied index
   public static final String OM_RATIS_SNAPSHOT_INDEX = "ratisSnapshotIndex";
+
+  // OM Http request parameter to be used while downloading DB checkpoint
+  // from OM leader to follower
+  public static final String OM_RATIS_SNAPSHOT_BEFORE_DB_CHECKPOINT =
+      "snapshotBeforeCheckpoint";
 }

hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/DBCheckpoint.java

@@ -55,4 +55,20 @@ public interface DBCheckpoint {
    */
   void cleanupCheckpoint() throws IOException;
 
+  /**
+   * Set the OM Ratis snapshot index corresponding to the OM DB checkpoint.
+   * The snapshot index is the latest snapshot index saved by ratis
+   * snapshots. It is not guaranteed to be the last ratis index applied to
+   * the OM DB state.
+   * @param omRatisSnapshotIndex the saved ratis snapshot index
+   */
+  void setRatisSnapshotIndex(long omRatisSnapshotIndex);
+
+  /**
+   * Get the OM Ratis snapshot index corresponding to the OM DB checkpoint.
+   * The ratis snapshot index indicates upto which index is definitely
+   * included in the DB checkpoint. It is not guaranteed to be the last ratis
+   * log index applied to the DB checkpoint.
+   */
+  long getRatisSnapshotIndex();
 }

hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/RocksDBCheckpoint.java

@@ -38,6 +38,7 @@ public class RocksDBCheckpoint implements DBCheckpoint {
   private long checkpointTimestamp = System.currentTimeMillis();
   private long latestSequenceNumber = -1;
   private long checkpointCreationTimeTaken = 0L;
+  private long ratisSnapshotIndex = 0L;
 
   public RocksDBCheckpoint(Path checkpointLocation) {
     this.checkpointLocation = checkpointLocation;
@@ -78,4 +79,14 @@ public void cleanupCheckpoint() throws IOException {
     LOG.debug("Cleaning up checkpoint at " + checkpointLocation.toString());
     FileUtils.deleteDirectory(checkpointLocation.toFile());
   }
+
+  @Override
+  public void setRatisSnapshotIndex(long omRatisSnapshotIndex) {
+    this.ratisSnapshotIndex = omRatisSnapshotIndex;
+  }
+
+  @Override
+  public long getRatisSnapshotIndex() {
+    return ratisSnapshotIndex;
+  }
 }

hadoop-hdds/common/src/main/resources/ozone-default.xml

@@ -1585,6 +1585,8 @@
       logs. If this is not set then default metadata dirs is used. A warning
       will be logged if this not set. Ideally, this should be mapped to a
       fast disk like an SSD.
+      If undefined, OM ratis storage dir will fallback to ozone.metadata.dirs.
+      This fallback approach is not recommended for production environments.
     </description>
   </property>
 
@@ -1703,6 +1705,45 @@
       .</description>
   </property>
 
+  <property>
+    <name>ozone.om.ratis.snapshot.dir</name>
+    <value/>
+    <tag>OZONE, OM, STORAGE, MANAGEMENT, RATIS</tag>
+    <description>This directory is used for storing OM's snapshot
+      related files like the ratisSnapshotIndex and DB checkpoint from leader
+      OM.
+      If undefined, OM snapshot dir will fallback to ozone.om.ratis.storage.dir.
+      This fallback approach is not recommended for production environments.
+    </description>
+  </property>
+  <property>
+    <name>ozone.om.snapshot.provider.socket.timeout</name>
+    <value>5000s</value>
+    <tag>OZONE, OM, HA, MANAGEMENT</tag>
+    <description>
+      Socket timeout for HTTP call made by OM Snapshot Provider to request
+      OM snapshot from OM Leader.
+    </description>
+  </property>
+  <property>
+    <name>ozone.om.snapshot.provider.connection.timeout</name>
+    <value>5000s</value>
+    <tag>OZONE, OM, HA, MANAGEMENT</tag>
+    <description>
+      Connection timeout for HTTP call made by OM Snapshot Provider to request
+      OM snapshot from OM Leader.
+    </description>
+  </property>
+  <property>
+    <name>ozone.om.snapshot.provider.request.timeout</name>
+    <value>5000ms</value>
+    <tag>OZONE, OM, HA, MANAGEMENT</tag>
+    <description>
+      Connection request timeout for HTTP call made by OM Snapshot Provider to
+      request OM snapshot from OM Leader.
+    </description>
+  </property>
+
   <property>
     <name>ozone.acl.authorizer.class</name>
     <value>org.apache.hadoop.ozone.security.acl.OzoneAccessAuthorizer</value>
@@ -2346,14 +2387,6 @@
       OM snapshot.
     </description>
   </property>
-  <property>
-    <name>recon.om.socket.timeout</name>
-    <value>5s</value>
-    <tag>OZONE, RECON, OM</tag>
-    <description>
-      Socket timeout for HTTP call made by Recon to request OM snapshot.
-    </description>
-  </property>
   <property>
     <name>recon.om.snapshot.task.initial.delay</name>
     <value>1m</value>

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rest/RestClient.java

@@ -99,6 +99,7 @@
 
 import static java.net.HttpURLConnection.HTTP_CREATED;
 import static java.net.HttpURLConnection.HTTP_OK;
+import static org.apache.hadoop.ozone.OzoneConsts.OZONE_OM_SERVICE_LIST_HTTP_ENDPOINT;
 
 /**
  * Ozone Client REST protocol implementation. It uses REST protocol to
@@ -190,7 +191,8 @@ private InetSocketAddress getOzoneRestServerAddress(
               " details on configuring Ozone.");
     }
 
-    HttpGet httpGet = new HttpGet("http://" + httpAddress + "/serviceList");
+    HttpGet httpGet = new HttpGet("http://" + httpAddress +
+        OZONE_OM_SERVICE_LIST_HTTP_ENDPOINT);
     HttpEntity entity = executeHttpRequest(httpGet);
     try {
       String serviceListJson = EntityUtils.toString(entity);

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/OmUtils.java

@@ -27,19 +27,23 @@
 import java.net.InetSocketAddress;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Optional;
 import java.util.zip.GZIPOutputStream;
 
+import com.google.common.base.Strings;
 import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
 import org.apache.commons.compress.archivers.tar.TarArchiveOutputStream;
 import org.apache.commons.compress.utils.IOUtils;
 import org.apache.commons.lang3.RandomStringUtils;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hdds.scm.ScmUtils;
+import org.apache.hadoop.hdds.scm.HddsServerUtil;
 import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.ozone.om.OMConfigKeys;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
@@ -48,7 +52,11 @@
 import static org.apache.hadoop.hdds.HddsUtils.getPortNumberFromConfigKeys;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_ADDRESS_KEY;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_BIND_HOST_DEFAULT;
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_HTTPS_ADDRESS_KEY;
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_HTTPS_BIND_HOST_KEY;
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_HTTPS_BIND_PORT_DEFAULT;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_HTTP_ADDRESS_KEY;
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_HTTP_BIND_HOST_KEY;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_HTTP_BIND_PORT_DEFAULT;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_NODES_KEY;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_PORT_DEFAULT;
@@ -366,4 +374,101 @@ private static void addFilesToArchive(String source, File file,
     }
   }
 
+  /**
+   * If a OM conf is only set with key suffixed with OM Node ID, return the
+   * set value.
+   * @return null if base conf key is set, otherwise the value set for
+   * key suffixed with Node ID.
+   */
+  public static String getConfSuffixedWithOMNodeId(Configuration conf,
+      String confKey, String omServiceID, String omNodeId) {
+    String confValue = conf.getTrimmed(confKey);
+    if (StringUtils.isNotEmpty(confValue)) {
+      return null;
+    }
+    String suffixedConfKey = OmUtils.addKeySuffixes(
+        confKey, omServiceID, omNodeId);
+    confValue = conf.getTrimmed(suffixedConfKey);
+    if (StringUtils.isNotEmpty(confValue)) {
+      return confValue;
+    }
+    return null;
+  }
+
+  /**
+   * Returns the http address of peer OM node.
+   * @param conf Configuration
+   * @param omNodeId peer OM node ID
+   * @param omNodeHostAddr peer OM node host address
+   * @return http address of peer OM node in the format <hostName>:<port>
+   */
+  public static String getHttpAddressForOMPeerNode(Configuration conf,
+      String omServiceId, String omNodeId, String omNodeHostAddr) {
+    final Optional<String> bindHost = getHostNameFromConfigKeys(conf,
+        addKeySuffixes(OZONE_OM_HTTP_BIND_HOST_KEY, omServiceId, omNodeId));
+
+    final Optional<Integer> addressPort = getPortNumberFromConfigKeys(conf,
+        addKeySuffixes(OZONE_OM_HTTP_ADDRESS_KEY, omServiceId, omNodeId));
+
+    final Optional<String> addressHost = getHostNameFromConfigKeys(conf,
+        addKeySuffixes(OZONE_OM_HTTP_ADDRESS_KEY, omServiceId, omNodeId));
+
+    String hostName = bindHost.orElse(addressHost.orElse(omNodeHostAddr));
+
+    return hostName + ":" + addressPort.orElse(OZONE_OM_HTTP_BIND_PORT_DEFAULT);
+  }
+
+  /**
+   * Returns the https address of peer OM node.
+   * @param conf Configuration
+   * @param omNodeId peer OM node ID
+   * @param omNodeHostAddr peer OM node host address
+   * @return https address of peer OM node in the format <hostName>:<port>
+   */
+  public static String getHttpsAddressForOMPeerNode(Configuration conf,
+      String omServiceId, String omNodeId, String omNodeHostAddr) {
+    final Optional<String> bindHost = getHostNameFromConfigKeys(conf,
+        addKeySuffixes(OZONE_OM_HTTPS_BIND_HOST_KEY, omServiceId, omNodeId));
+
+    final Optional<Integer> addressPort = getPortNumberFromConfigKeys(conf,
+        addKeySuffixes(OZONE_OM_HTTPS_ADDRESS_KEY, omServiceId, omNodeId));
+
+    final Optional<String> addressHost = getHostNameFromConfigKeys(conf,
+        addKeySuffixes(OZONE_OM_HTTPS_ADDRESS_KEY, omServiceId, omNodeId));
+
+    String hostName = bindHost.orElse(addressHost.orElse(omNodeHostAddr));
+
+    return hostName + ":" +
+        addressPort.orElse(OZONE_OM_HTTPS_BIND_PORT_DEFAULT);
+  }
+
+  /**
+   * Get the local directory where ratis logs will be stored.
+   */
+  public static String getOMRatisDirectory(Configuration conf) {
+    String storageDir = conf.get(OMConfigKeys.OZONE_OM_RATIS_STORAGE_DIR);
+
+    if (Strings.isNullOrEmpty(storageDir)) {
+      storageDir = HddsServerUtil.getDefaultRatisDirectory(conf);
+    }
+    return storageDir;
+  }
+
+  public static String getOMRatisSnapshotDirectory(Configuration conf) {
+    String snapshotDir = conf.get(OMConfigKeys.OZONE_OM_RATIS_SNAPSHOT_DIR);
+
+    if (Strings.isNullOrEmpty(snapshotDir)) {
+      snapshotDir = Paths.get(getOMRatisDirectory(conf),
+          "snapshot").toString();
+    }
+    return snapshotDir;
+  }
+
+  public static File createOMDir(String dirPath) {
+    File dirFile = new File(dirPath);
+    if (!dirFile.exists() && !dirFile.mkdirs()) {
+      throw new IllegalArgumentException("Unable to create path: " + dirFile);
+    }
+    return dirFile;
+  }
 }

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OMConfigKeys.java

@@ -184,6 +184,27 @@ private OMConfigKeys() {
       OZONE_OM_RATIS_SERVER_ROLE_CHECK_INTERVAL_DEFAULT
       = TimeDuration.valueOf(15, TimeUnit.SECONDS);
 
+  // OM SnapshotProvider configurations
+  public static final String OZONE_OM_RATIS_SNAPSHOT_DIR =
+      "ozone.om.ratis.snapshot.dir";
+  public static final String OZONE_OM_SNAPSHOT_PROVIDER_SOCKET_TIMEOUT_KEY =
+      "ozone.om.snapshot.provider.socket.timeout";
+  public static final TimeDuration
+      OZONE_OM_SNAPSHOT_PROVIDER_SOCKET_TIMEOUT_DEFAULT =
+      TimeDuration.valueOf(5000, TimeUnit.MILLISECONDS);
+
+  public static final String OZONE_OM_SNAPSHOT_PROVIDER_CONNECTION_TIMEOUT_KEY =
+      "ozone.om.snapshot.provider.connection.timeout";
+  public static final TimeDuration
+      OZONE_OM_SNAPSHOT_PROVIDER_CONNECTION_TIMEOUT_DEFAULT =
+      TimeDuration.valueOf(5000, TimeUnit.MILLISECONDS);
+
+  public static final String OZONE_OM_SNAPSHOT_PROVIDER_REQUEST_TIMEOUT_KEY =
+      "ozone.om.snapshot.provider.request.timeout";
+  public static final TimeDuration
+      OZONE_OM_SNAPSHOT_PROVIDER_REQUEST_TIMEOUT_DEFAULT =
+      TimeDuration.valueOf(5000, TimeUnit.MILLISECONDS);
+
   public static final String OZONE_OM_KERBEROS_KEYTAB_FILE_KEY = "ozone.om."
       + "kerberos.keytab.file";
   public static final String OZONE_OM_KERBEROS_PRINCIPAL_KEY = "ozone.om"

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneHAClusterImpl.java

@@ -200,18 +200,17 @@ private Map<String, OzoneManager> createOMService() throws IOException,
             // Set nodeId
             String nodeId = nodeIdBaseStr + i;
             conf.set(OMConfigKeys.OZONE_OM_NODE_ID_KEY, nodeId);
+            // Set the OM http(s) address to null so that the cluster picks
+            // up the address set with service ID and node ID in initHAConfig
+            conf.set(OMConfigKeys.OZONE_OM_HTTP_ADDRESS_KEY, "");
+            conf.set(OMConfigKeys.OZONE_OM_HTTPS_ADDRESS_KEY, "");
 
             // Set metadata/DB dir base path
             String metaDirPath = path + "/" + nodeId;
             conf.set(OZONE_METADATA_DIRS, metaDirPath);
             OMStorage omStore = new OMStorage(conf);
             initializeOmStorage(omStore);
 
-            // Set HTTP address to the rpc port + 2
-            int httpPort = basePort + (6*i) - 4;
-            conf.set(OMConfigKeys.OZONE_OM_HTTP_ADDRESS_KEY,
-                "127.0.0.1:" + httpPort);
-
             OzoneManager om = OzoneManager.createOm(null, conf);
             om.setCertClient(certClient);
             omMap.put(nodeId, om);
@@ -261,11 +260,16 @@ private void initHAConfig(int basePort) throws IOException {
         omNodesKeyValue.append(",").append(omNodeId);
         String omAddrKey = OmUtils.addKeySuffixes(
             OMConfigKeys.OZONE_OM_ADDRESS_KEY, omServiceId, omNodeId);
+        String omHttpAddrKey = OmUtils.addKeySuffixes(
+            OMConfigKeys.OZONE_OM_HTTP_ADDRESS_KEY, omServiceId, omNodeId);
+        String omHttpsAddrKey = OmUtils.addKeySuffixes(
+            OMConfigKeys.OZONE_OM_HTTPS_ADDRESS_KEY, omServiceId, omNodeId);
         String omRatisPortKey = OmUtils.addKeySuffixes(
             OMConfigKeys.OZONE_OM_RATIS_PORT_KEY, omServiceId, omNodeId);
 
         conf.set(omAddrKey, "127.0.0.1:" + port);
-        // Reserve port+2 for OMs HTTP server
+        conf.set(omHttpAddrKey, "127.0.0.1:" + (port + 2));
+        conf.set(omHttpsAddrKey, "127.0.0.1:" + (port + 3));
         conf.setInt(omRatisPortKey, port + 4);
       }