diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/conf/OzoneConfiguration.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/conf/OzoneConfiguration.java index f35073c4728b..91415c34ecce 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/conf/OzoneConfiguration.java +++ b/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) }); } } diff --git a/hadoop-hdds/common/src/main/resources/ozone-default.xml b/hadoop-hdds/common/src/main/resources/ozone-default.xml index d23473ec8567..a1c87877753e 100644 --- a/hadoop-hdds/common/src/main/resources/ozone-default.xml +++ b/hadoop-hdds/common/src/main/resources/ozone-default.xml @@ -1579,9 +1579,9 @@ hdds.datanode.replication.work.dir DATANODE - 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. + 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). diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ContainerUtils.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ContainerUtils.java index e0480735e5b3..81a6935098af 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ContainerUtils.java +++ b/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); + } + } } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/ContainerPacker.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/ContainerPacker.java index 8308c23866b8..a7c7f5ad20fd 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/ContainerPacker.java +++ b/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 { * file but returned). */ byte[] unpackContainerData(Container container, - InputStream inputStream) + InputStream inputStream, Path tmpDir, Path destContainerDir) throws IOException; /** diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java index 99e931f833e7..4b8f0be1e3e4 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java +++ b/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); diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/volume/StorageVolume.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/volume/StorageVolume.java index 561708b852cd..a5cddc175eaf 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/volume/StorageVolume.java +++ b/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; } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java index 7412e766d0c0..bd58b7253b99 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java +++ b/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 packer) throws IOException { + ContainerPacker packer) + throws IOException { + HddsVolume hddsVolume = containerData.getVolume(); + String idDir = VersionedDatanodeFeatures.ScmHA.chooseContainerPathID( + hddsVolume, hddsVolume.getClusterID()); + long containerId = containerData.getContainerID(); + Path destContainerDir = + Paths.get(KeyValueContainerLocationUtil.getBaseContainerLocation( + hddsVolume.getHddsRootDir().toString(), idDir, containerId)); + 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,31 +537,28 @@ public void importContainerData(InputStream input, KeyValueContainerData originalContainerData = (KeyValueContainerData) ContainerDataYaml .readContainer(descriptorContent); - - - containerData.setState(originalContainerData.getState()); - containerData - .setContainerDBType(originalContainerData.getContainerDBType()); - containerData.setSchemaVersion(originalContainerData.getSchemaVersion()); - - //rewriting the yaml file with new checksum calculation. - update(originalContainerData.getMetadata(), true); - - if (containerData.getSchemaVersion().equals(OzoneConsts.SCHEMA_V3)) { - // load metadata from received dump files before we try to parse kv - BlockUtils.loadKVContainerDataFromFiles(containerData, config); + importContainerData(originalContainerData); + } catch (Exception ex) { + // clean data under tmp directory + try { + Path containerUntarDir = tmpDir.resolve(String.valueOf(containerId)); + if (containerUntarDir.toFile().exists()) { + FileUtils.deleteDirectory(containerUntarDir.toFile()); + } + } catch (Exception deleteex) { + LOG.error( + "Can not cleanup container directory under {} for container {}", + tmpDir, containerId, deleteex); } - //fill in memory stat counter (keycount, byte usage) - KeyValueContainerUtil.parseKVContainerData(containerData, config); - - } catch (Exception ex) { + // Throw exception for existed containers if (ex instanceof StorageContainerException && ((StorageContainerException) ex).getResult() == CONTAINER_ALREADY_EXISTS) { throw ex; } - //delete all the temporary data in case of any exception. + + // delete all other temporary data in case of any exception. try { if (containerData.getSchemaVersion() != null && containerData.getSchemaVersion().equals(OzoneConsts.SCHEMA_V3)) { @@ -570,8 +571,7 @@ public void importContainerData(InputStream input, } catch (Exception deleteex) { LOG.error( "Can not cleanup destination directories after a container import" - + " error (cid" + - containerData.getContainerID() + ")", deleteex); + + " error (cid: {}", containerId, deleteex); } throw ex; } finally { @@ -579,6 +579,25 @@ public void importContainerData(InputStream input, } } + public void importContainerData(KeyValueContainerData originalContainerData) + throws IOException { + containerData.setState(originalContainerData.getState()); + containerData + .setContainerDBType(originalContainerData.getContainerDBType()); + containerData.setSchemaVersion(originalContainerData.getSchemaVersion()); + + //rewriting the yaml file with new checksum calculation. + update(originalContainerData.getMetadata(), true); + + if (containerData.getSchemaVersion().equals(OzoneConsts.SCHEMA_V3)) { + // load metadata from received dump files before we try to parse kv + BlockUtils.loadKVContainerDataFromFiles(containerData, config); + } + + //fill in memory stat counter (keycount, byte usage) + KeyValueContainerUtil.parseKVContainerData(containerData, config); + } + @Override public void exportContainerData(OutputStream destination, ContainerPacker packer) throws IOException { diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java index d003af86be9a..7a4cc5807e4a 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java @@ -71,7 +71,6 @@ import org.apache.hadoop.ozone.container.common.statemachine.DatanodeConfiguration; import org.apache.hadoop.ozone.container.common.transport.server.ratis.DispatcherContext; import org.apache.hadoop.ozone.container.common.transport.server.ratis.DispatcherContext.WriteChunkStage; -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; @@ -381,13 +380,11 @@ 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()); String idDir = VersionedDatanodeFeatures.ScmHA.chooseContainerPathID( containerVolume, clusterId); container.populatePathFields(idDir, containerVolume); @@ -1014,8 +1011,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 +1021,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; diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/TarContainerPacker.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/TarContainerPacker.java index 4c851979edeb..3d8c445a84fe 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/TarContainerPacker.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/TarContainerPacker.java @@ -28,15 +28,19 @@ import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; -import java.util.Objects; +import java.nio.file.StandardCopyOption; import java.util.stream.Stream; +import java.util.Objects; import com.google.common.annotations.VisibleForTesting; + import org.apache.hadoop.hdds.HddsUtils; +import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException; import org.apache.hadoop.ozone.OzoneConsts; import org.apache.hadoop.ozone.container.common.interfaces.Container; import org.apache.hadoop.ozone.container.common.interfaces.ContainerPacker; + import org.apache.commons.compress.archivers.ArchiveEntry; import org.apache.commons.compress.archivers.ArchiveInputStream; import org.apache.commons.compress.archivers.ArchiveOutputStream; @@ -44,10 +48,13 @@ import org.apache.commons.compress.archivers.tar.TarArchiveOutputStream; import org.apache.commons.compress.compressors.CompressorException; import org.apache.commons.compress.compressors.CompressorStreamFactory; +import org.apache.commons.io.FileUtils; import org.apache.commons.io.IOUtils; +import org.apache.hadoop.ozone.container.keyvalue.helpers.KeyValueContainerLocationUtil; import org.apache.hadoop.ozone.container.metadata.DatanodeStoreSchemaThreeImpl; import static java.util.stream.Collectors.toList; +import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.CONTAINER_ALREADY_EXISTS; import static org.apache.hadoop.ozone.OzoneConsts.SCHEMA_V3; /** @@ -83,48 +90,35 @@ public TarContainerPacker(String compression) { */ @Override public byte[] unpackContainerData(Container container, - InputStream input) + InputStream input, Path tmpDir, Path destContainerDir) throws IOException { - byte[] descriptorFileContent = null; KeyValueContainerData containerData = container.getContainerData(); - Path dbRoot = getDbPath(containerData); - Path chunksRoot = Paths.get(containerData.getChunksPath()); + long containerId = containerData.getContainerID(); - try (InputStream decompressed = decompress(input); - ArchiveInputStream archiveInput = untar(decompressed)) { + Path containerUntarDir = tmpDir.resolve(String.valueOf(containerId)); + if (containerUntarDir.toFile().exists()) { + FileUtils.deleteDirectory(containerUntarDir.toFile()); + } - ArchiveEntry entry = archiveInput.getNextEntry(); - while (entry != null) { - String name = entry.getName(); - long size = entry.getSize(); - if (name.startsWith(DB_DIR_NAME + "/")) { - Path destinationPath = dbRoot - .resolve(name.substring(DB_DIR_NAME.length() + 1)); - extractEntry(entry, archiveInput, size, dbRoot, - destinationPath); - } else if (name.startsWith(CHUNKS_DIR_NAME + "/")) { - Path destinationPath = chunksRoot - .resolve(name.substring(CHUNKS_DIR_NAME.length() + 1)); - extractEntry(entry, archiveInput, size, chunksRoot, - destinationPath); - } else if (CONTAINER_FILE_NAME.equals(name)) { - //Don't do anything. Container file should be unpacked in a - //separated step by unpackContainerDescriptor call. - descriptorFileContent = readEntry(archiveInput, size); - } else { - throw new IllegalArgumentException( - "Unknown entry in the tar file: " + "" + name); - } - entry = archiveInput.getNextEntry(); - } - return descriptorFileContent; + Path dbRoot = getDbPath(containerUntarDir, containerData); + Path chunksRoot = getChunkPath(containerUntarDir, containerData); + byte[] descriptorFileContent = innerUnpack(input, dbRoot, chunksRoot); - } catch (CompressorException e) { - throw new IOException( - "Can't uncompress the given container: " + container - .getContainerData().getContainerID(), - e); + if (!Files.exists(destContainerDir)) { + Files.createDirectories(destContainerDir); + } + if (FileUtils.isEmptyDirectory(destContainerDir.toFile())) { + Files.move(containerUntarDir, destContainerDir, + StandardCopyOption.ATOMIC_MOVE, + StandardCopyOption.REPLACE_EXISTING); + } else { + String errorMessage = "Container " + containerId + + " unpack failed because ContainerFile " + + destContainerDir.toAbsolutePath() + " already exists"; + throw new StorageContainerException(errorMessage, + CONTAINER_ALREADY_EXISTS); } + return descriptorFileContent; } private void extractEntry(ArchiveEntry entry, InputStream input, long size, @@ -223,6 +217,33 @@ public static Path getDbPath(KeyValueContainerData containerData) { } } + public static Path getDbPath(Path baseDir, + KeyValueContainerData containerData) { + if (baseDir.toAbsolutePath().toString().equals( + containerData.getContainerPath())) { + return getDbPath(containerData); + } + Path containerPath = Paths.get(containerData.getContainerPath()); + Path dbPath = Paths.get(containerData.getDbFile().getPath()); + Path relativePath = containerPath.relativize(dbPath); + + if (containerData.getSchemaVersion().equals(SCHEMA_V3)) { + Path metadataDir = KeyValueContainerLocationUtil.getContainerMetaDataPath( + baseDir.toString()).toPath(); + return DatanodeStoreSchemaThreeImpl.getDumpDir(metadataDir.toFile()) + .toPath(); + } else { + return baseDir.resolve(relativePath); + } + } + + public static Path getChunkPath(Path baseDir, + KeyValueContainerData containerData) { + Path chunkDir = KeyValueContainerLocationUtil.getChunksLocationPath( + baseDir.toString()).toPath(); + return chunkDir; + } + private byte[] readEntry(InputStream input, final long size) throws IOException { ByteArrayOutputStream output = new ByteArrayOutputStream(); @@ -289,4 +310,40 @@ OutputStream compress(OutputStream output) .createCompressorOutputStream(compression, output); } + private byte[] innerUnpack(InputStream input, Path dbRoot, Path chunksRoot) + throws IOException { + byte[] descriptorFileContent = null; + try (InputStream decompressed = decompress(input); + ArchiveInputStream archiveInput = untar(decompressed)) { + ArchiveEntry entry = archiveInput.getNextEntry(); + while (entry != null) { + String name = entry.getName(); + long size = entry.getSize(); + if (name.startsWith(DB_DIR_NAME + "/")) { + Path destinationPath = dbRoot + .resolve(name.substring(DB_DIR_NAME.length() + 1)); + extractEntry(entry, archiveInput, size, dbRoot, + destinationPath); + } else if (name.startsWith(CHUNKS_DIR_NAME + "/")) { + Path destinationPath = chunksRoot + .resolve(name.substring(CHUNKS_DIR_NAME.length() + 1)); + extractEntry(entry, archiveInput, size, chunksRoot, + destinationPath); + } else if (CONTAINER_FILE_NAME.equals(name)) { + //Don't do anything. Container file should be unpacked in a + //separated step by unpackContainerDescriptor call. + descriptorFileContent = readEntry(archiveInput, size); + } else { + throw new IllegalArgumentException( + "Unknown entry in the tar file: " + "" + name); + } + entry = archiveInput.getNextEntry(); + } + return descriptorFileContent; + + } catch (CompressorException e) { + throw new IOException("Can't uncompress to dbRoot: " + dbRoot + + ", chunksRoot: " + chunksRoot, e); + } + } } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/helpers/KeyValueContainerLocationUtil.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/helpers/KeyValueContainerLocationUtil.java index 13a5a6981155..591efe04e2e5 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/helpers/KeyValueContainerLocationUtil.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/helpers/KeyValueContainerLocationUtil.java @@ -45,15 +45,22 @@ private KeyValueContainerLocationUtil() { public static File getContainerMetaDataPath(String hddsVolumeDir, String clusterId, long containerId) { - String containerMetaDataPath = - getBaseContainerLocation(hddsVolumeDir, clusterId, - containerId); - containerMetaDataPath = containerMetaDataPath + File.separator + + return getContainerMetaDataPath( + getBaseContainerLocation(hddsVolumeDir, clusterId, containerId)); + } + + /** + * Returns Container Metadata Location. + * @param containerBaseDir Base container dir + * @return containerMetadata Path to container metadata location where + * .container file will be stored. + */ + public static File getContainerMetaDataPath(String containerBaseDir) { + String containerMetaDataPath = containerBaseDir + File.separator + OzoneConsts.CONTAINER_META_PATH; return new File(containerMetaDataPath); } - /** * Returns Container Chunks Location. * @param baseDir @@ -63,8 +70,17 @@ public static File getContainerMetaDataPath(String hddsVolumeDir, */ public static File getChunksLocationPath(String baseDir, String clusterId, long containerId) { - String chunksPath = - getBaseContainerLocation(baseDir, clusterId, containerId) + return getChunksLocationPath( + getBaseContainerLocation(baseDir, clusterId, containerId)); + } + + /** + * Returns Container Chunks Location. + * @param containerBaseDir + * @return chunksPath + */ + public static File getChunksLocationPath(String containerBaseDir) { + String chunksPath = containerBaseDir + File.separator + OzoneConsts.STORAGE_DIR_CHUNKS; return new File(chunksPath); } @@ -76,7 +92,7 @@ public static File getChunksLocationPath(String baseDir, String clusterId, * @param containerId * @return base directory for container. */ - private static String getBaseContainerLocation(String hddsVolumeDir, + public static String getBaseContainerLocation(String hddsVolumeDir, String clusterId, long containerId) { Preconditions.checkNotNull(hddsVolumeDir, "Base Directory cannot be null"); @@ -111,7 +127,12 @@ public static File getContainerDBFile(KeyValueContainerData containerData) { return new File(containerData.getVolume().getDbParentDir(), OzoneConsts.CONTAINER_DB_NAME); } - return new File(containerData.getMetadataPath(), - containerData.getContainerID() + OzoneConsts.DN_CONTAINER_DB); + return getContainerDBFile(containerData.getMetadataPath(), containerData); + } + + public static File getContainerDBFile(String baseDir, + KeyValueContainerData containerData) { + return new File(baseDir, containerData.getContainerID() + + OzoneConsts.DN_CONTAINER_DB); } } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/ContainerController.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/ContainerController.java index 171303dc0b4d..4087483d723f 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/ContainerController.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/ContainerController.java @@ -156,8 +156,7 @@ public void closeContainer(final long containerId) throws IOException { public Container importContainer( final ContainerData containerData, final InputStream rawContainerStream, - final TarContainerPacker packer) - throws IOException { + final TarContainerPacker packer) throws IOException { return handlers.get(containerData.getContainerType()) .importContainer(containerData, rawContainerStream, packer); } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ContainerDownloader.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ContainerDownloader.java index cd4ebffd6163..45c70f8e72b5 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ContainerDownloader.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ContainerDownloader.java @@ -34,6 +34,6 @@ public interface ContainerDownloader extends Closeable { Path getContainerDataFromReplicas(long containerId, - List sources); + List sources, Path downloadDir); } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/DownloadAndImportReplicator.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/DownloadAndImportReplicator.java index c97bd27e93ab..c1d99c9148dd 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/DownloadAndImportReplicator.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/DownloadAndImportReplicator.java @@ -21,13 +21,22 @@ import java.io.IOException; import java.nio.file.Files; import java.nio.file.Path; +import java.nio.file.Paths; import java.util.List; +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.ozone.container.common.impl.ContainerData; +import org.apache.hadoop.hdds.scm.ScmConfigKeys; import org.apache.hadoop.ozone.container.common.impl.ContainerDataYaml; import org.apache.hadoop.ozone.container.common.impl.ContainerSet; import org.apache.hadoop.ozone.container.common.interfaces.Container; +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.MutableVolumeSet; +import org.apache.hadoop.ozone.container.common.volume.RoundRobinVolumeChoosingPolicy; +import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainerData; import org.apache.hadoop.ozone.container.keyvalue.TarContainerPacker; import org.apache.hadoop.ozone.container.ozoneimpl.ContainerController; import org.apache.hadoop.ozone.container.replication.ReplicationTask.Status; @@ -35,6 +44,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_DATANODE_VOLUME_CHOOSING_POLICY; + /** * Default replication implementation. *

@@ -46,36 +57,59 @@ public class DownloadAndImportReplicator implements ContainerReplicator { public static final Logger LOG = LoggerFactory.getLogger(DownloadAndImportReplicator.class); - private final ContainerSet containerSet; + public static final String CONTAINER_COPY_DIR = "container-copy"; + public static final String CONTAINER_COPY_TMP_DIR = "tmp"; + private final ContainerSet containerSet; private final ContainerController controller; - private final ContainerDownloader downloader; - private final TarContainerPacker packer; + private final MutableVolumeSet volumeSet; + private final VolumeChoosingPolicy volumeChoosingPolicy; + private final long containerSize; public DownloadAndImportReplicator( + ConfigurationSource conf, ContainerSet containerSet, ContainerController controller, ContainerDownloader downloader, - TarContainerPacker packer) { + TarContainerPacker packer, + MutableVolumeSet volumeSet) { this.containerSet = containerSet; this.controller = controller; this.downloader = downloader; this.packer = packer; + 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); + } - public void importContainer(long containerID, Path tarFilePath) - throws IOException { + public void importContainer(long containerID, Path tarFilePath, + HddsVolume hddsVolume) throws IOException { + + HddsVolume targetVolume = hddsVolume; + if (targetVolume == null) { + targetVolume = chooseNextVolume(); + } + KeyValueContainerData originalContainerData; try { - ContainerData originalContainerData; - try (FileInputStream tempContainerTarStream = new FileInputStream( + try (FileInputStream tmpContainerTarStream = new FileInputStream( tarFilePath.toFile())) { byte[] containerDescriptorYaml = - packer.unpackContainerDescriptor(tempContainerTarStream); - originalContainerData = ContainerDataYaml.readContainer( - containerDescriptorYaml); + packer.unpackContainerDescriptor(tmpContainerTarStream); + originalContainerData = (KeyValueContainerData) ContainerDataYaml + .readContainer(containerDescriptorYaml); } + originalContainerData.setVolume(targetVolume); try (FileInputStream tempContainerTarStream = new FileInputStream( tarFilePath.toFile())) { @@ -105,26 +139,46 @@ public void replicate(ReplicationTask task) { LOG.info("Starting replication of container {} from {}", containerID, sourceDatanodes); - // Wait for the download. This thread pool is limiting the parallel - // downloads, so it's ok to block here and wait for the full download. - Path path = - downloader.getContainerDataFromReplicas(containerID, sourceDatanodes); - if (path == null) { - task.setStatus(Status.FAILED); - } else { - try { - long bytes = Files.size(path); - LOG.info("Container {} is downloaded with size {}, starting to import.", - containerID, bytes); - task.setTransferredBytes(bytes); - - importContainer(containerID, path); - LOG.info("Container {} is replicated successfully", containerID); - task.setStatus(Status.DONE); - } catch (IOException e) { - LOG.error("Container {} replication was unsuccessful.", containerID, e); + try { + HddsVolume targetVolume = chooseNextVolume(); + // Wait for the download. This thread pool is limiting the parallel + // downloads, so it's ok to block here and wait for the full download. + Path tarFilePath = + downloader.getContainerDataFromReplicas(containerID, sourceDatanodes, + getUntarDirectory(targetVolume)); + if (tarFilePath == null) { task.setStatus(Status.FAILED); + return; } + long bytes = Files.size(tarFilePath); + LOG.info("Container {} is downloaded with size {}, starting to import.", + containerID, bytes); + task.setTransferredBytes(bytes); + + importContainer(containerID, tarFilePath, targetVolume); + + LOG.info("Container {} is replicated successfully", containerID); + task.setStatus(Status.DONE); + } catch (IOException e) { + LOG.error("Container {} replication was unsuccessful.", containerID, e); + task.setStatus(Status.FAILED); } } + + private HddsVolume chooseNextVolume() throws IOException { + // Choose volume that can hold both container in tmp and dest directory + return volumeChoosingPolicy.chooseVolume( + StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()), + containerSize * 2); + } + + public static Path getUntarDirectory(HddsVolume hddsVolume) + throws IOException { + return Paths.get(hddsVolume.getVolumeRootDir()) + .resolve(CONTAINER_COPY_TMP_DIR).resolve(CONTAINER_COPY_DIR); + } + + private List getHddsVolumesList() { + return StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()); + } } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/GrpcReplicationClient.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/GrpcReplicationClient.java index 580b725fc440..b8c2c6fbb473 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/GrpcReplicationClient.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/GrpcReplicationClient.java @@ -38,6 +38,7 @@ import org.apache.hadoop.ozone.OzoneConsts; import com.google.common.base.Preconditions; +import org.apache.hadoop.ozone.container.common.helpers.ContainerUtils; import org.apache.ratis.thirdparty.io.grpc.ManagedChannel; import org.apache.ratis.thirdparty.io.grpc.netty.GrpcSslContexts; import org.apache.ratis.thirdparty.io.grpc.netty.NettyChannelBuilder; @@ -107,8 +108,8 @@ public CompletableFuture download(long containerId) { CompletableFuture response = new CompletableFuture<>(); - Path destinationPath = - getWorkingDirectory().resolve("container-" + containerId + ".tar.gz"); + Path destinationPath = getWorkingDirectory() + .resolve(ContainerUtils.getContainerTarGzName(containerId)); client.download(request, new StreamDownloader(containerId, response, destinationPath)); diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/SimpleContainerDownloader.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/SimpleContainerDownloader.java index e46cd7c1b19a..329fd64fa2f4 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/SimpleContainerDownloader.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/SimpleContainerDownloader.java @@ -32,12 +32,13 @@ import org.apache.hadoop.hdds.protocol.DatanodeDetails.Port.Name; 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.slf4j.Logger; import org.slf4j.LoggerFactory; +import static org.apache.hadoop.ozone.container.replication.DownloadAndImportReplicator.CONTAINER_COPY_DIR; + /** * Simple ContainerDownloaderImplementation to download the missing container @@ -51,22 +52,12 @@ public class SimpleContainerDownloader implements ContainerDownloader { public static final Logger LOG = LoggerFactory.getLogger(SimpleContainerDownloader.class); - private final Path workingDirectory; private final SecurityConfig securityConfig; private final CertificateClient certClient; private final String compression; - public SimpleContainerDownloader(ConfigurationSource conf, - CertificateClient certClient) { - 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); - } + public SimpleContainerDownloader( + ConfigurationSource conf, CertificateClient certClient) { securityConfig = new SecurityConfig(conf); this.certClient = certClient; this.compression = CopyContainerCompression.getConf(conf).toString(); @@ -74,7 +65,13 @@ public SimpleContainerDownloader(ConfigurationSource conf, @Override public Path getContainerDataFromReplicas( - long containerId, List sourceDatanodes) { + long containerId, List sourceDatanodes, + Path downloadDir) { + + if (downloadDir == null) { + downloadDir = Paths.get(System.getProperty("java.io.tmpdir")) + .resolve(CONTAINER_COPY_DIR); + } final List shuffledDatanodes = shuffleDatanodes(sourceDatanodes); @@ -82,7 +79,7 @@ public Path getContainerDataFromReplicas( for (DatanodeDetails datanode : shuffledDatanodes) { try { CompletableFuture result = - downloadContainer(containerId, datanode); + downloadContainer(containerId, datanode, downloadDir); return result.get(); } catch (ExecutionException | IOException e) { LOG.error("Error on replicating container: {} from {}/{}", containerId, @@ -115,12 +112,14 @@ protected List shuffleDatanodes( @VisibleForTesting protected CompletableFuture downloadContainer( - long containerId, DatanodeDetails datanode) throws IOException { + long containerId, DatanodeDetails datanode, Path downloadDir) + throws IOException { CompletableFuture result; GrpcReplicationClient grpcReplicationClient = new GrpcReplicationClient(datanode.getIpAddress(), datanode.getPort(Name.REPLICATION).getValue(), - workingDirectory, securityConfig, certClient, compression); + downloadDir, securityConfig, certClient, compression); + result = grpcReplicationClient.download(containerId) .whenComplete((r, ex) -> { try { diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/helpers/TestContainerUtils.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/helpers/TestContainerUtils.java index 6ae0461fc904..51b929eb2b45 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/helpers/TestContainerUtils.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/helpers/TestContainerUtils.java @@ -24,6 +24,7 @@ import org.apache.hadoop.ozone.common.ChunkBuffer; import org.junit.jupiter.api.Test; +import java.io.IOException; import java.nio.ByteBuffer; import static java.nio.charset.StandardCharsets.UTF_8; @@ -57,4 +58,13 @@ public void redactsDataBuffers() { assertEquals("", dataBuffers.getBuffers(0).toString(UTF_8)); } + @Test + public void testTarGzName() throws IOException { + long containerId = 100; + String tarGzName = "container-100.tar.gz"; + assertEquals(tarGzName, ContainerUtils.getContainerTarGzName(containerId)); + + assertEquals(containerId, + ContainerUtils.retrieveContainerIdFromTarGzName(tarGzName)); + } } diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestTarContainerPacker.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestTarContainerPacker.java index 2d834716917f..9e16fb88bc7e 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestTarContainerPacker.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestTarContainerPacker.java @@ -136,15 +136,22 @@ private static void initDir(Path path) throws IOException { } private KeyValueContainerData createContainer(Path dir) throws IOException { + return createContainer(dir, true); + } + + private KeyValueContainerData createContainer(Path dir, boolean createDir) + throws IOException { long id = CONTAINER_ID.getAndIncrement(); - Path containerDir = dir.resolve("container" + id); + Path containerDir = dir.resolve(String.valueOf(id)); Path dbDir = containerDir.resolve("db"); - Path dataDir = containerDir.resolve("data"); + Path dataDir = containerDir.resolve("chunks"); Path metaDir = containerDir.resolve("metadata"); - Files.createDirectories(metaDir); - Files.createDirectories(dbDir); - Files.createDirectories(dataDir); + if (createDir) { + Files.createDirectories(metaDir); + Files.createDirectories(dbDir); + Files.createDirectories(dataDir); + } KeyValueContainerData containerData = new KeyValueContainerData( id, layout, @@ -211,7 +218,7 @@ public void pack() throws IOException, CompressorException { } KeyValueContainerData destinationContainerData = - createContainer(DEST_CONTAINER_ROOT); + createContainer(DEST_CONTAINER_ROOT, false); KeyValueContainer destinationContainer = new KeyValueContainer(destinationContainerData, conf); @@ -221,7 +228,10 @@ public void pack() throws IOException, CompressorException { //unpackContainerData try (FileInputStream input = new FileInputStream(targetFile.toFile())) { descriptor = - new String(packer.unpackContainerData(destinationContainer, input), + new String(packer.unpackContainerData(destinationContainer, input, + TEMP_DIR, + DEST_CONTAINER_ROOT.resolve(String.valueOf( + destinationContainer.getContainerData().getContainerID()))), UTF_8); } @@ -316,9 +326,10 @@ public void unpackContainerDataWithInvalidRelativeChunkFilePath() private KeyValueContainerData unpackContainerData(File containerFile) throws IOException { try (FileInputStream input = new FileInputStream(containerFile)) { - KeyValueContainerData data = createContainer(DEST_CONTAINER_ROOT); + KeyValueContainerData data = createContainer(DEST_CONTAINER_ROOT, false); KeyValueContainer container = new KeyValueContainer(data, conf); - packer.unpackContainerData(container, input); + packer.unpackContainerData(container, input, TEMP_DIR, + DEST_CONTAINER_ROOT.resolve(String.valueOf(data.getContainerID()))); return data; } } diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestReplicationSupervisor.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestReplicationSupervisor.java index d9770a14e5d5..2ea47f597262 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestReplicationSupervisor.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestReplicationSupervisor.java @@ -18,8 +18,10 @@ package org.apache.hadoop.ozone.container.replication; +import java.io.IOException; import java.nio.file.Path; import java.nio.file.Paths; +import java.util.Collections; import java.time.Instant; import java.time.ZoneId; import java.util.List; @@ -36,6 +38,8 @@ import org.apache.hadoop.metrics2.impl.MetricsCollectorImpl; import org.apache.hadoop.ozone.container.common.impl.ContainerLayoutVersion; import org.apache.hadoop.ozone.container.common.impl.ContainerSet; +import org.apache.hadoop.ozone.container.common.volume.HddsVolume; +import org.apache.hadoop.ozone.container.common.volume.MutableVolumeSet; import org.apache.hadoop.ozone.container.common.statemachine.DatanodeStateMachine; import org.apache.hadoop.ozone.container.common.statemachine.StateContext; import org.apache.hadoop.ozone.container.keyvalue.ContainerLayoutTestInfo; @@ -245,21 +249,30 @@ public void slowDownload() { } @Test - public void testDownloadAndImportReplicatorFailure() { + public void testDownloadAndImportReplicatorFailure() throws IOException { ReplicationSupervisor supervisor = new ReplicationSupervisor(set, context, mutableReplicator, newDirectExecutorService(), clock); + OzoneConfiguration conf = new OzoneConfiguration(); // Mock to fetch an exception in the importContainer method. SimpleContainerDownloader moc = Mockito.mock(SimpleContainerDownloader.class); Path res = Paths.get("file:/tmp/no-such-file"); Mockito.when( - moc.getContainerDataFromReplicas(Mockito.anyLong(), Mockito.anyList())) + moc.getContainerDataFromReplicas(Mockito.anyLong(), Mockito.anyList(), + Mockito.any(Path.class))) .thenReturn(res); + final String testDir = GenericTestUtils.getTempPath( + TestReplicationSupervisor.class.getSimpleName() + + "-" + UUID.randomUUID().toString()); + MutableVolumeSet volumeSet = Mockito.mock(MutableVolumeSet.class); + Mockito.when(volumeSet.getVolumesList()) + .thenReturn(Collections.singletonList( + new HddsVolume.Builder(testDir).conf(conf).build())); ContainerReplicator replicatorFactory = - new DownloadAndImportReplicator(set, null, moc, null); + new DownloadAndImportReplicator(conf, set, null, moc, null, volumeSet); replicatorRef.set(replicatorFactory); diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestSimpleContainerDownloader.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestSimpleContainerDownloader.java index bc83bf9bcb3c..42d44a7c2399 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestSimpleContainerDownloader.java +++ b/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,8 +34,14 @@ 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.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; import org.junit.jupiter.api.Assertions; -import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; /** @@ -40,6 +49,9 @@ */ public class TestSimpleContainerDownloader { + @Rule + public final TemporaryFolder tempDir = new TemporaryFolder(); + private static final String SUCCESS_PATH = "downloaded"; @Test @@ -53,7 +65,8 @@ public void testGetContainerDataFromReplicasHappyPath() throws Exception { //WHEN final Path result = - downloader.getContainerDataFromReplicas(1L, datanodes); + downloader.getContainerDataFromReplicas(1L, datanodes, + tempDir.newFolder().toPath()); //THEN Assertions.assertEquals(datanodes.get(0).getUuidString(), @@ -72,7 +85,8 @@ public void testGetContainerDataFromReplicasDirectFailure() //WHEN final Path result = - downloader.getContainerDataFromReplicas(1L, datanodes); + downloader.getContainerDataFromReplicas(1L, datanodes, + tempDir.newFolder().toPath()); //THEN //first datanode is failed, second worked @@ -91,7 +105,8 @@ public void testGetContainerDataFromReplicasAsyncFailure() throws Exception { //WHEN final Path result = - downloader.getContainerDataFromReplicas(1L, datanodes); + downloader.getContainerDataFromReplicas(1L, datanodes, + tempDir.newFolder().toPath()); //THEN //first datanode is failed, second worked @@ -105,7 +120,7 @@ public void testGetContainerDataFromReplicasAsyncFailure() throws Exception { @Test @Timeout(10) public void testRandomSelection() - throws ExecutionException, InterruptedException { + throws ExecutionException, InterruptedException, IOException { //GIVEN final List datanodes = createDatanodes(); @@ -115,7 +130,7 @@ public void testRandomSelection() @Override protected CompletableFuture downloadContainer( - long containerId, DatanodeDetails datanode + long containerId, DatanodeDetails datanode, Path downloadPath ) { //download is always successful. return CompletableFuture @@ -126,7 +141,8 @@ protected CompletableFuture downloadContainer( //WHEN executed, THEN at least once the second datanode should be //returned. for (int i = 0; i < 10000; i++) { - Path path = downloader.getContainerDataFromReplicas(1L, datanodes); + Path path = downloader.getContainerDataFromReplicas(1L, datanodes, + tempDir.newFolder().toPath()); if (path.toString().equals(datanodes.get(1).getUuidString())) { return; } @@ -167,8 +183,7 @@ protected List shuffleDatanodes( @Override protected CompletableFuture downloadContainer( - long containerId, - DatanodeDetails datanode + long containerId, DatanodeDetails datanode, Path downloadPath ) { if (datanodes.contains(datanode)) { @@ -197,4 +212,21 @@ private List 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; + } } diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/upgrade/TestDatanodeUpgradeToScmHA.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/upgrade/TestDatanodeUpgradeToScmHA.java index ec0d9bc335b9..1e01d0808043 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/upgrade/TestDatanodeUpgradeToScmHA.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/upgrade/TestDatanodeUpgradeToScmHA.java @@ -33,6 +33,7 @@ import org.apache.hadoop.ozone.container.common.DatanodeLayoutStorage; import org.apache.hadoop.ozone.container.common.SCMTestUtils; import org.apache.hadoop.ozone.container.common.ScmTestMock; +import org.apache.hadoop.ozone.container.common.helpers.ContainerUtils; import org.apache.hadoop.ozone.container.common.statemachine.DatanodeStateMachine; import org.apache.hadoop.ozone.container.common.statemachine.EndpointStateMachine; import org.apache.hadoop.ozone.container.common.states.endpoint.VersionEndpointTask; @@ -666,15 +667,17 @@ public File exportContainer(long containerId) throws Exception { */ public void importContainer(long containerID, File source) throws Exception { DownloadAndImportReplicator replicator = - new DownloadAndImportReplicator(dsm.getContainer().getContainerSet(), + new DownloadAndImportReplicator(dsm.getConf(), + dsm.getContainer().getContainerSet(), dsm.getContainer().getController(), new SimpleContainerDownloader(conf, null), - new TarContainerPacker()); + new TarContainerPacker(), dsm.getContainer().getVolumeSet()); - File tempFile = tempFolder.newFile(); + File tempFile = tempFolder.newFile( + ContainerUtils.getContainerTarGzName(containerID)); Files.copy(source.toPath(), tempFile.toPath(), StandardCopyOption.REPLACE_EXISTING); - replicator.importContainer(containerID, tempFile.toPath()); + replicator.importContainer(containerID, tempFile.toPath(), null); } public void dispatchRequest( diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneClusterImpl.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneClusterImpl.java index d0c897b488c5..2e0429dccc19 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneClusterImpl.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneClusterImpl.java @@ -855,8 +855,6 @@ protected List 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(); diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/ozoneimpl/TestOzoneContainerWithTLS.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/ozoneimpl/TestOzoneContainerWithTLS.java index 861f7ca8c4c4..61418b722d0a 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/ozoneimpl/TestOzoneContainerWithTLS.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/ozoneimpl/TestOzoneContainerWithTLS.java @@ -272,7 +272,7 @@ public void testContainerDownload() throws Exception { SimpleContainerDownloader downloader = new SimpleContainerDownloader(conf, caClient); Path file = downloader.getContainerDataFromReplicas( - containerId, sourceDatanodes); + containerId, sourceDatanodes, null); downloader.close(); Assert.assertNull(file); Assert.assertTrue(logCapture.getOutput().contains( @@ -309,7 +309,8 @@ public void testContainerDownload() throws Exception { for (Long cId : containerIdList) { downloader = new SimpleContainerDownloader(conf, caClient); try { - file = downloader.getContainerDataFromReplicas(cId, sourceDatanodes); + file = downloader.getContainerDataFromReplicas(cId, sourceDatanodes, + null); downloader.close(); Assert.assertNotNull(file); } finally { diff --git a/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/freon/ClosedContainerReplicator.java b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/freon/ClosedContainerReplicator.java index 0f0a95f71f06..1c0ba2a80c49 100644 --- a/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/freon/ClosedContainerReplicator.java +++ b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/freon/ClosedContainerReplicator.java @@ -203,10 +203,10 @@ private void initializeReplicationSupervisor(ConfigurationSource conf) new ContainerController(containerSet, handlers); ContainerReplicator replicator = - new DownloadAndImportReplicator(containerSet, + new DownloadAndImportReplicator(conf, containerSet, controller, new SimpleContainerDownloader(conf, null), - new TarContainerPacker()); + new TarContainerPacker(), null); ReplicationServer.ReplicationConfig replicationConfig = conf.getObject(ReplicationServer.ReplicationConfig.class);