Skip to content

HDDS-4475.Extend DatanodeChunkGenerator to write all on all pipelines… #1600

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
elek merged 5 commits into from
Jan 12, 2021
Merged

HDDS-4475.Extend DatanodeChunkGenerator to write all on all pipelines… #1600

Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
603e3e8
HDDS-4475.Extend DatanodeChunkGenerator to write all on all pipelines…
sadanand48 Nov 18, 2020
e8224b0
Addressed review comments. added datanodes argument
sadanand48 Nov 30, 2020
8b08703
addressed comments
sadanand48 Dec 23, 2020
9e084ee
addressed comments
sadanand48 Jan 7, 2021
8a0d797
fixed typos & addressed comment
sadanand48 Jan 12, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
150 changes: 112 additions & 38 deletions hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/freon/DatanodeChunkGenerator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,12 +16,17 @@
*/
package org.apache.hadoop.ozone.freon;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;

import org.apache.hadoop.hdds.cli.HddsVersionProvider;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChecksumData;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChecksumType;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChunkInfo;
Expand Down Expand Up @@ -75,7 +80,12 @@ public class DatanodeChunkGenerator extends BaseFreonGenerator implements
description = "Pipeline to use. By default the first RATIS/THREE "
+ "pipeline will be used.",
defaultValue = "")
private String pipelineId;
private String pipelineIds;

@Option(names = {"-d", "--datanodes"},
description = "Datanodes to use. ",

@elek elek Jan 11, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
description = "Datanodes to use. ",
description = "Datanodes to use. Test will write to all the existing pipelines which this datanode is member of.",

defaultValue = "")
private String datanodes;

private XceiverClientSpi xceiverClientSpi;

Expand All @@ -84,54 +94,47 @@ public class DatanodeChunkGenerator extends BaseFreonGenerator implements
private ByteString dataToWrite;
private ChecksumData checksumProtobuf;


@Override
public Void call() throws Exception {

init();

OzoneConfiguration ozoneConf = createOzoneConfiguration();
if (OzoneSecurityUtil.isSecurityEnabled(ozoneConf)) {
throw new IllegalArgumentException(
"Datanode chunk generator is not supported in secure environment");
}

try (StorageContainerLocationProtocol scmLocationClient =
createStorageContainerLocationClient(ozoneConf)) {
List<Pipeline> pipelines = scmLocationClient.listPipelines();
Pipeline pipeline;
if (pipelineId != null && pipelineId.length() > 0) {
pipeline = pipelines.stream()
.filter(p -> p.getId().toString().equals(pipelineId))
.findFirst()
.orElseThrow(() -> new IllegalArgumentException(
"Pipeline ID is defined, but there is no such pipeline: "
+ pipelineId));

} else {
pipeline = pipelines.stream()
.filter(p -> p.getFactor() == ReplicationFactor.THREE)
.findFirst()
.orElseThrow(() -> new IllegalArgumentException(
"Pipeline ID is NOT defined, and no pipeline " +
"has been found with factor=THREE"));
LOG.info("Using pipeline {}", pipeline.getId());
}

try (XceiverClientManager xceiverClientManager =
new XceiverClientManager(ozoneConf)) {
xceiverClientSpi = xceiverClientManager.acquireClient(pipeline);
List<String> pipelines = Arrays.asList(pipelineIds.split(","));

timer = getMetrics().timer("chunk-write");
List<String> datanodeHosts = Arrays.asList(this.datanodes.split(","));

byte[] data = RandomStringUtils.randomAscii(chunkSize)
.getBytes(StandardCharsets.UTF_8);
pipelines = getPipelinesFromDatanodes(ozoneConf, pipelines, datanodeHosts);

dataToWrite = ByteString.copyFrom(data);

Checksum checksum = new Checksum(ChecksumType.CRC32, chunkSize);
checksumProtobuf = checksum.computeChecksum(data).getProtoBufMessage();

runTests(this::writeChunk);
try (StorageContainerLocationProtocol scmLocationClient =
createStorageContainerLocationClient(ozoneConf)) {
List<Pipeline> pipelinesFromSCM = scmLocationClient.listPipelines();
Pipeline temp;
if (!arePipelinesOrDatanodesProvided()) {
//default behaviour if no arguments provided
init();
temp = pipelinesFromSCM.stream()
.filter(p -> p.getFactor() == ReplicationFactor.THREE)
.findFirst()
.orElseThrow(() -> new IllegalArgumentException(
"Pipeline ID is NOT defined, and no pipeline " +
"has been found with factor=THREE"));
LOG.info("Using pipeline {}", temp.getId());
runTest(ozoneConf, temp);
} else {
List<CompletableFuture> futures = new ArrayList<>();
for(String pipelineId:pipelines){
futures.add(CompletableFuture.runAsync(()->writeOnPipeline(
ozoneConf, pipelinesFromSCM, pipelineId)));
}
CompletableFuture.allOf(futures.toArray(
new CompletableFuture[pipelines.size()]))
.join();
}
} finally {
if (xceiverClientSpi != null) {
Expand All @@ -141,6 +144,70 @@ public Void call() throws Exception {
return null;
}

private void writeOnPipeline(OzoneConfiguration ozoneConf,
List<Pipeline> pipelinesFromSCM, String pipelineId) {
Pipeline temp;
init();
LOG.info("Writing block for pipeline:" + pipelineId);
temp = pipelinesFromSCM.stream()
.filter(p -> p.getId().toString().equals("PipelineID=" + pipelineId))
.findFirst()
.orElseThrow(() -> new IllegalArgumentException(
"Pipeline ID is defined, but there is no such pipeline: "
+ pipelineId));
try {
runTest(ozoneConf, temp);
} catch (IOException e) {
LOG.error("Could not write chunks");
}
}

private boolean arePipelinesOrDatanodesProvided() {
return !(pipelineIds.equals("") && datanodes.equals(""));
}

private List<String> getPipelinesFromDatanodes(OzoneConfiguration ozoneConf,
List<String> pipelines, List<String> dns) throws IOException {
if(!datanodes.equals("")) {
try (StorageContainerLocationProtocol scmLocationClient =
createStorageContainerLocationClient(ozoneConf)) {
List<Pipeline> pipelineList = scmLocationClient.listPipelines();
pipelines = new ArrayList<String>();
for (Pipeline p : pipelineList) {
for (DatanodeDetails dn : p.getNodes()) {
if (dns.contains(dn.getHostName())) {
if(!pipelines.contains(p.getId().toString().substring(11))) {
pipelines.add(p.getId().toString().substring(11));
}
}
}
LOG.debug("Pipeline list size:" + pipelines.size());
}
}
}
return pipelines;
}

private void runTest(OzoneConfiguration ozoneConf, Pipeline pipeline)
throws IOException {
try (XceiverClientManager xceiverClientManager =
new XceiverClientManager(ozoneConf)) {
xceiverClientSpi = xceiverClientManager.acquireClient(pipeline);

timer = getMetrics().timer("chunk-write");

byte[] data = RandomStringUtils.randomAscii(chunkSize)
.getBytes(StandardCharsets.UTF_8);

dataToWrite = ByteString.copyFrom(data);

Checksum checksum = new Checksum(ChecksumType.CRC32, chunkSize);
checksumProtobuf = checksum.computeChecksum(data).getProtoBufMessage();

runTests(this::writeChunk);
}
}

private void writeChunk(long stepNo)
throws Exception {

Expand All @@ -165,7 +232,15 @@ private void writeChunk(long stepNo)
.setChunkData(chunkInfo)
.setData(dataToWrite);

String id = xceiverClientSpi.getPipeline().getFirstNode().getUuidString();
sendWriteChunkRequest(blockId, writeChunkRequest,
xceiverClientSpi.getPipeline().getFirstNode());

}

private void sendWriteChunkRequest(DatanodeBlockID blockId,
WriteChunkRequestProto.Builder writeChunkRequest,
DatanodeDetails datanodeDetails) throws Exception {
String id = datanodeDetails.getUuidString();

ContainerCommandRequestProto.Builder builder =
ContainerCommandRequestProto
Expand All @@ -188,7 +263,6 @@ private void writeChunk(long stepNo)
}
return null;
});

}

}