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

@@ -1554,9 +1554,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. Default directory which is used
+      during the container replication betweeen datanodes. Replication work will
+      be spread to different volumes.
     </description>
   </property>
 

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

@@ -168,7 +168,8 @@ public DatanodeStateMachine(DatanodeDetails datanodeDetails,
     ContainerReplicator replicator =
         new DownloadAndImportReplicator(container.getContainerSet(),
             container.getController(),
-            new SimpleContainerDownloader(conf, dnCertClient),
+            new SimpleContainerDownloader(conf, dnCertClient,
+                container.getVolumeSet()),
             new TarContainerPacker());
 
     replicatorMetrics = new MeasuredReplicator(replicator);

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/SimpleContainerDownloader.java

@@ -28,16 +28,24 @@
 import java.util.concurrent.ExecutionException;
 
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.conf.StorageUnit;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails.Port.Name;
+import org.apache.hadoop.hdds.scm.ScmConfigKeys;
 import org.apache.hadoop.hdds.security.x509.SecurityConfig;
 import org.apache.hadoop.hdds.security.x509.certificate.client.CertificateClient;
-import org.apache.hadoop.ozone.OzoneConfigKeys;
 
 import com.google.common.annotations.VisibleForTesting;
+import org.apache.hadoop.ozone.container.common.interfaces.VolumeChoosingPolicy;
+import org.apache.hadoop.ozone.container.common.utils.StorageVolumeUtil;
+import org.apache.hadoop.ozone.container.common.volume.HddsVolume;
+import org.apache.hadoop.ozone.container.common.volume.RoundRobinVolumeChoosingPolicy;
+import org.apache.hadoop.ozone.container.common.volume.VolumeSet;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_DATANODE_VOLUME_CHOOSING_POLICY;
+
 /**
  * Simple ContainerDownloaderImplementation to download the missing container
  * from the first available datanode.
@@ -50,24 +58,33 @@ public class SimpleContainerDownloader implements ContainerDownloader {
   private static final Logger LOG =
       LoggerFactory.getLogger(SimpleContainerDownloader.class);
 
-  private final Path workingDirectory;
+  public static final String CONTAINER_COPY_DIR = "container-copy";
+
+  private ConfigurationSource conf;
   private final SecurityConfig securityConfig;
   private final CertificateClient certClient;
+  private final VolumeSet volumeSet;
+  private VolumeChoosingPolicy volumeChoosingPolicy = null;
+  private long containerSize;
 
   public SimpleContainerDownloader(
-      ConfigurationSource conf, CertificateClient certClient) {
+      ConfigurationSource conf, CertificateClient certClient,
+      VolumeSet volumeSet) {
 
-    String workDirString =
-        conf.get(OzoneConfigKeys.OZONE_CONTAINER_COPY_WORKDIR);
-
-    if (workDirString == null) {
-      workingDirectory = Paths.get(System.getProperty("java.io.tmpdir"))
-          .resolve("container-copy");
-    } else {
-      workingDirectory = Paths.get(workDirString);
-    }
+    this.conf = conf;
     securityConfig = new SecurityConfig(conf);
     this.certClient = certClient;
+    this.volumeSet = volumeSet;
+    try {
+      this.volumeChoosingPolicy = conf.getClass(
+          HDDS_DATANODE_VOLUME_CHOOSING_POLICY, RoundRobinVolumeChoosingPolicy
+              .class, VolumeChoosingPolicy.class).newInstance();
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+    this.containerSize = (long) conf.getStorageSize(
+        ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE,
+        ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE_DEFAULT, StorageUnit.BYTES);
   }
 
   @Override
@@ -118,7 +135,7 @@ protected CompletableFuture<Path> downloadContainer(
     GrpcReplicationClient grpcReplicationClient =
         new GrpcReplicationClient(datanode.getIpAddress(),
             datanode.getPort(Name.REPLICATION).getValue(),
-            workingDirectory, securityConfig, certClient);
+            getWorkingDirectory(), securityConfig, certClient);
     result = grpcReplicationClient.download(containerId)
         .whenComplete((r, ex) -> {
           try {
@@ -135,4 +152,21 @@ protected CompletableFuture<Path> downloadContainer(
   public void close() {
     // noop
   }
+
+  public Path getWorkingDirectory() {
+    Path defaultWorkingDirectory =
+        Paths.get(System.getProperty("java.io.tmpdir"));
+    try {
+      // Use containerSize * 2 to store source and dest file
+      HddsVolume volume = volumeChoosingPolicy.chooseVolume(
+          StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()),
+          containerSize * 2);
+      return Paths.get(volume.getStorageDir().getParent()).resolve("tmp")
+          .resolve(CONTAINER_COPY_DIR);
+    } catch (IOException e) {
+      LOG.error("Exception when spreading copy directory, using default " +
+          "working directory {}", defaultWorkingDirectory.toAbsolutePath(), e);
+    }
+    return defaultWorkingDirectory;
+  }
 }

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestSimpleContainerDownloader.java

@@ -18,11 +18,14 @@
 
 package org.apache.hadoop.ozone.container.replication;
 
+import java.io.File;
+import java.io.IOException;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 
@@ -31,14 +34,23 @@
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.MockDatanodeDetails;
 
+import org.apache.hadoop.hdds.scm.ScmConfigKeys;
+import org.apache.hadoop.ozone.container.common.volume.MutableVolumeSet;
+import org.apache.hadoop.ozone.container.common.volume.StorageVolume;
+import org.apache.hadoop.ozone.container.common.volume.VolumeSet;
 import org.junit.Assert;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
 
 /**
  * Test SimpleContainerDownloader.
  */
 public class TestSimpleContainerDownloader {
 
+  @Rule
+  public final TemporaryFolder tempDir = new TemporaryFolder();
+
   private static final String SUCCESS_PATH = "downloaded";
 
   @Test
@@ -106,7 +118,7 @@ public void testRandomSelection()
     final List<DatanodeDetails> datanodes = createDatanodes();
 
     SimpleContainerDownloader downloader =
-        new SimpleContainerDownloader(new OzoneConfiguration(), null) {
+        new SimpleContainerDownloader(new OzoneConfiguration(), null, null) {
 
           @Override
           protected CompletableFuture<Path> downloadContainer(
@@ -133,6 +145,29 @@ protected CompletableFuture<Path> downloadContainer(
             + "used.");
   }
 
+  @Test
+  public void testGetWorkingDirectory()
+      throws Exception {
+
+    //GIVEN
+    List<DatanodeDetails> datanodes = createDatanodes();
+
+    // Spread container copy directory
+    OzoneConfiguration conf = new OzoneConfiguration();
+    VolumeSet volumeSet = getVolumeSet(datanodes.get(0), conf);
+
+    SimpleContainerDownloader downloader =
+        new SimpleContainerDownloader(conf, null, volumeSet);
+
+    Path firstDi = downloader.getWorkingDirectory();
+    Path secondDi = downloader.getWorkingDirectory();
+
+    Assert.assertNotEquals(firstDi,
+        Paths.get(System.getProperty("java.io.tmpdir"))
+            .resolve(SimpleContainerDownloader.CONTAINER_COPY_DIR));
+    Assert.assertNotEquals(firstDi, secondDi);
+  }
+
   /**
    * Creates downloader which fails with datanodes in the arguments.
    *
@@ -149,7 +184,7 @@ private SimpleContainerDownloader createDownloaderWithPredefinedFailures(
     final List<DatanodeDetails> datanodes =
         Arrays.asList(failedDatanodes);
 
-    return new SimpleContainerDownloader(conf, null) {
+    return new SimpleContainerDownloader(conf, null, null) {
 
       //for retry testing we use predictable list of datanodes.
       @Override
@@ -192,4 +227,21 @@ private List<DatanodeDetails> createDatanodes() {
     datanodes.add(MockDatanodeDetails.randomDatanodeDetails());
     return datanodes;
   }
+
+  private VolumeSet getVolumeSet(DatanodeDetails datanodeDetails,
+      OzoneConfiguration conf) throws IOException {
+    String clusterId = UUID.randomUUID().toString();
+    int volumeNum = 3;
+    File[] hddsVolumeDirs = new File[volumeNum];
+    StringBuilder hddsDirs = new StringBuilder();
+    for (int i = 0; i < volumeNum; i++) {
+      hddsVolumeDirs[i] = tempDir.newFolder();
+      hddsDirs.append(hddsVolumeDirs[i]).append(",");
+    }
+    conf.set(ScmConfigKeys.HDDS_DATANODE_DIR_KEY, hddsDirs.toString());
+    VolumeSet hddsVolumeSet = new MutableVolumeSet(
+        datanodeDetails.getUuidString(), clusterId, conf, null,
+        StorageVolume.VolumeType.DATA_VOLUME, null);
+    return hddsVolumeSet;
+  }
 }

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/upgrade/TestDatanodeUpgradeToScmHA.java

@@ -668,7 +668,7 @@ public void importContainer(long containerID, File source) throws Exception {
     DownloadAndImportReplicator replicator =
         new DownloadAndImportReplicator(dsm.getContainer().getContainerSet(),
             dsm.getContainer().getController(),
-            new SimpleContainerDownloader(conf, null),
+            new SimpleContainerDownloader(conf, null, null),
             new TarContainerPacker());
 
     File tempFile = tempFolder.newFile();

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

@@ -838,8 +838,6 @@ protected List<HddsDatanodeService> createHddsDatanodes(
             reservedSpaceString);
         dnConf.set(OzoneConfigKeys.DFS_CONTAINER_RATIS_DATANODE_STORAGE_DIR,
             ratisDir.toString());
-        dnConf.set(OzoneConfigKeys.OZONE_CONTAINER_COPY_WORKDIR,
-            workDir.toString());
         if (reconServer != null) {
           OzoneStorageContainerManager reconScm =
               reconServer.getReconStorageContainerManager();

hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/freon/ClosedContainerReplicator.java

@@ -200,7 +200,7 @@ private void initializeReplicationSupervisor(ConfigurationSource conf)
     ContainerReplicator replicator =
         new DownloadAndImportReplicator(containerSet,
             controller,
-            new SimpleContainerDownloader(conf, null),
+            new SimpleContainerDownloader(conf, null, null),
             new TarContainerPacker());
 
     supervisor = new ReplicationSupervisor(containerSet, replicator, 10);