HDDS-7083. Spread container-copy directories

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/conf/OzoneConfiguration.java

@@ -46,6 +46,7 @@
 import org.apache.ratis.server.RaftServerConfigKeys;
 
 import static org.apache.hadoop.hdds.ratis.RatisHelper.HDDS_DATANODE_RATIS_PREFIX_KEY;
+import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_CONTAINER_COPY_WORKDIR;
 
 /**
  * Configuration for ozone.
@@ -308,7 +309,9 @@ private static void addDeprecatedKeys() {
         new DeprecationDelta("dfs.datanode.keytab.file",
             DFSConfigKeysLegacy.DFS_DATANODE_KERBEROS_KEYTAB_FILE_KEY),
         new DeprecationDelta("ozone.scm.chunk.layout",
-            ScmConfigKeys.OZONE_SCM_CONTAINER_LAYOUT_KEY)
+            ScmConfigKeys.OZONE_SCM_CONTAINER_LAYOUT_KEY),
+        new DeprecationDelta("hdds.datanode.replication.work.dir",
+            OZONE_CONTAINER_COPY_WORKDIR)
     });
   }
 }

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

@@ -1579,9 +1579,9 @@
   <property>
     <name>hdds.datanode.replication.work.dir</name>
     <tag>DATANODE</tag>
-    <description>Temporary which is used during the container replication
-      betweeen datanodes. Should have enough space to store multiple container
-      (in compressed format), but doesn't require fast io access such as SSD.
+    <description>This configuration is deprecated. Temporary sub directory under
+      each hdds.datanode.dir will be used during the container replication between
+      datanodes to save the downloaded container(in compressed format).
     </description>
   </property>
 

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ContainerUtils.java

@@ -33,6 +33,8 @@
 import java.nio.file.Paths;
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 import org.apache.commons.codec.digest.DigestUtils;
 import org.apache.hadoop.fs.FileAlreadyExistsException;
@@ -278,4 +280,22 @@ public static long getContainerID(File containerBaseDir) {
     return Long.parseLong(containerBaseDir.getName());
   }
 
+  public static String getContainerTarGzName(long containerId) {
+    return "container-" + containerId + ".tar.gz";
+  }
+
+  public static long retrieveContainerIdFromTarGzName(String tarGzName)
+      throws IOException {
+    assert tarGzName != null;
+    Pattern pattern = Pattern.compile("container-(\\d+).tar.gz");
+    // Now create matcher object.
+    Matcher m = pattern.matcher(tarGzName);
+
+    if (m.find()) {
+      return Long.parseLong(m.group(1));
+    } else {
+      throw new IOException("Illegal container tar gz file " +
+          tarGzName);
+    }
+  }
 }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/ContainerPacker.java

@@ -21,6 +21,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.nio.file.Path;
 
 import org.apache.hadoop.ozone.container.common.impl.ContainerData;
 
@@ -39,7 +40,7 @@ public interface ContainerPacker<CONTAINERDATA extends ContainerData> {
    * file but returned).
    */
   byte[] unpackContainerData(Container<CONTAINERDATA> container,
-      InputStream inputStream)
+      InputStream inputStream, Path tmpDir, Path destContainerDir)
       throws IOException;
 
   /**

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java

@@ -172,10 +172,10 @@ public DatanodeStateMachine(DatanodeDetails datanodeDetails,
     nextHB = new AtomicLong(Time.monotonicNow());
 
     ContainerReplicator replicator =
-        new DownloadAndImportReplicator(container.getContainerSet(),
+        new DownloadAndImportReplicator(conf, container.getContainerSet(),
             container.getController(),
             new SimpleContainerDownloader(conf, dnCertClient),
-            new TarContainerPacker());
+            new TarContainerPacker(), container.getVolumeSet());
 
     replicatorMetrics = new MeasuredReplicator(replicator);
 

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/utils/HddsVolumeUtil.java

@@ -26,6 +26,7 @@
 import org.apache.hadoop.ozone.container.keyvalue.helpers.BlockUtils;
 import org.apache.hadoop.ozone.container.metadata.DatanodeStore;
 import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.File;
 import java.io.IOException;
@@ -40,6 +41,9 @@
  */
 public final class HddsVolumeUtil {
 
+  private static final Logger LOG =
+      LoggerFactory.getLogger(HddsVolumeUtil.class);
+
   // Private constructor for Utility class. Unused.
   private HddsVolumeUtil() {
   }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/volume/StorageVolume.java

@@ -368,6 +368,10 @@ public StorageType getStorageType() {
     }
   }
 
+  public String getVolumeRootDir() {
+    return volumeInfo != null ? volumeInfo.getRootDir() : null;
+  }
+
   public long getCapacity() {
     return volumeInfo != null ? volumeInfo.getCapacity() : 0;
   }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java

@@ -23,6 +23,8 @@
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.nio.file.StandardCopyOption;
 import java.time.Instant;
 import java.util.Collections;
@@ -56,6 +58,7 @@
 import org.apache.hadoop.ozone.container.keyvalue.helpers.BlockUtils;
 import org.apache.hadoop.ozone.container.keyvalue.helpers.KeyValueContainerLocationUtil;
 import org.apache.hadoop.ozone.container.keyvalue.helpers.KeyValueContainerUtil;
+import org.apache.hadoop.ozone.container.replication.DownloadAndImportReplicator;
 import org.apache.hadoop.ozone.container.upgrade.VersionedDatanodeFeatures;
 import org.apache.hadoop.util.DiskChecker.DiskOutOfSpaceException;
 
@@ -507,21 +510,22 @@ public void updateDeleteTransactionId(long deleteTransactionId) {
 
   @Override
   public void importContainerData(InputStream input,
-      ContainerPacker<KeyValueContainerData> packer) throws IOException {
+      ContainerPacker<KeyValueContainerData> packer)
+      throws IOException {
+    HddsVolume hddsVolume = containerData.getVolume();
+    String idDir = VersionedDatanodeFeatures.ScmHA.chooseContainerPathID(
+        hddsVolume, hddsVolume.getClusterID());
+    Path destContainerDir =
+        Paths.get(KeyValueContainerLocationUtil.getBaseContainerLocation(
+            hddsVolume.getHddsRootDir().toString(), idDir,
+            containerData.getContainerID()));
+    Path tmpDir = DownloadAndImportReplicator.getUntarDirectory(hddsVolume);
     writeLock();
     try {
-      if (getContainerFile().exists()) {
-        String errorMessage = String.format(
-            "Can't import container (cid=%d) data to a specific location"
-                + " as the container descriptor (%s) has already been exist.",
-            getContainerData().getContainerID(),
-            getContainerFile().getAbsolutePath());
-        throw new StorageContainerException(errorMessage,
-            CONTAINER_ALREADY_EXISTS);
-      }
       //copy the values from the input stream to the final destination
       // directory.
-      byte[] descriptorContent = packer.unpackContainerData(this, input);
+      byte[] descriptorContent = packer.unpackContainerData(this, input, tmpDir,
+          destContainerDir);
 
       Preconditions.checkNotNull(descriptorContent,
           "Container descriptor is missing from the container archive: "
@@ -533,8 +537,16 @@ public void importContainerData(InputStream input,
       KeyValueContainerData originalContainerData =
           (KeyValueContainerData) ContainerDataYaml
               .readContainer(descriptorContent);
+      importContainerData(originalContainerData);
+    } finally {
+      writeUnlock();
+    }
+  }
 
-
+  public void importContainerData(KeyValueContainerData originalContainerData)
+      throws IOException {
+    writeLock();
+    try {
       containerData.setState(originalContainerData.getState());
       containerData
           .setContainerDBType(originalContainerData.getContainerDBType());

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java

@@ -381,13 +381,16 @@ ContainerCommandResponseProto handleCreateContainer(
     return getSuccessResponse(request);
   }
 
-  private void populateContainerPathFields(KeyValueContainer container)
-      throws IOException {
+  private void populateContainerPathFields(KeyValueContainer container,
+      HddsVolume hddsVolume) throws IOException {
     volumeSet.readLock();
+    HddsVolume containerVolume = hddsVolume;
     try {
-      HddsVolume containerVolume = volumeChoosingPolicy.chooseVolume(
-          StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()),
-          container.getContainerData().getMaxSize());
+      if (hddsVolume == null) {
+        containerVolume = volumeChoosingPolicy.chooseVolume(
+            StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()),
+            container.getContainerData().getMaxSize());
+      }
       String idDir = VersionedDatanodeFeatures.ScmHA.chooseContainerPathID(
               containerVolume, clusterId);
       container.populatePathFields(idDir, containerVolume);
@@ -1014,8 +1017,7 @@ private void checkContainerOpen(KeyValueContainer kvContainer)
   @Override
   public Container importContainer(ContainerData originalContainerData,
       final InputStream rawContainerStream,
-      final TarContainerPacker packer)
-      throws IOException {
+      final TarContainerPacker packer) throws IOException {
     Preconditions.checkState(originalContainerData instanceof
         KeyValueContainerData, "Should be KeyValueContainerData instance");
 
@@ -1025,7 +1027,8 @@ public Container importContainer(ContainerData originalContainerData,
     KeyValueContainer container = new KeyValueContainer(containerData,
         conf);
 
-    populateContainerPathFields(container);
+    HddsVolume targetVolume = originalContainerData.getVolume();
+    populateContainerPathFields(container, targetVolume);
     container.importContainerData(rawContainerStream, packer);
     sendICR(container);
     return container;