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

[sadanand48]

What changes were proposed in this pull request?

Currently, DatanodeChunkGenerator takes a single pipeline as a parameter. This will allow passing a list of pipelines as comma-separated by their pipeline ids and the load will be generated on the dns of the provided pipelines.

What is the link to the Apache JIRA

https://issues.apache.org/jira/browse/HDDS-4475

How was this patch tested?

Tested on docker.

[bshashikant]

@sadanand48 , as discuseed offline. Can you make the write chunk tests to run concurrently rather than sequentially?

Also, can we also add support for multiple datanodes along with pipeline Ids?

[bshashikant]

@elek , can you plz have a look at this?

[elek]

This will allow passing a list of pipelines as comma-separated by their pipeline ids and the load will be generated on the dns of the provided pipelines.

Thanks for the patch @sadanand48. I don't fully understand this line. Can you please explain the goals in more details?

I have one guess: the goal is to define the pipeline with the help of a datanode host name.

Today we have two options:

1. If pipelineId is defined, that will be used
2. If pipline is not defined, we check the SCM and will use the first Ratis/THREE open pipeline

In both cases the first datanode in the pipeline will be used to save chunk.

If I understood well, this patch would like to offer a third option: if the datanode is set, the list of the pipelines will be further restricted to the pipelines where the specified datanode is a member.

If this is the goal, it should be possible with adjusting the filter conditions:

        temp = pipelinesFromSCM.stream()
              .filter(p -> p.getFactor() == ReplicationFactor.THREE)
              .filter(p -> datanodeHosts.size() ==0 || pipelineContainsDatanode(p, datanodeHosts))
              .findFirst()
              .orElseThrow(() -> new IllegalArgumentException(
                  "Pipeline ID is NOT defined, and no pipeline " +
                      "has been found with factor=THREE"));

(see the second filter)

I agree with @bshashikant, we shouldn't remove the logic of the parallel executions for this specific case.

(But please let me know if I misunderstood something)

[sadanand48]

Thanks @elek for the comments . The current tool takes only 1 pipelineID as parameter and generates chunks on this pipeline.If any arguments not provided, it will by default select the RATIS THREE pipeline and write to it.
The goal here is to

1. Pass multiple pipelineID's at once and generate write chunk requests on them .
2. Pass datanode/datanodes and select those pipelines of which these datanodes are part of and write to them. I will use the second filter for that like you suggested.
   The default behaviour remains the same.

[elek]

Thanks to explain it, now I got it.

In that case you should have additional modification to use the current loop:

You should either select the first one OR the pipelines assigned to the datanode.

Change private XceiverClientSpi xceiverClientSpi to private List<XceiverClientSpi> clients to have clients to all pipelines.

And in writeChunk select a random one instead of using the single one.

(You should also initialize the full list of clients and close them...)

Also: how would you like to guarantee that if the selected datanode is the leader of bot pipelines?

[elek]

(You can ping me offline if it was not clear, need help or if my understanding is till not right ;-) )

[elek]

Thanks the update @sadanand48

This part seems to be suspicious:

 179   │   private void writeOnPipeline(OzoneConfiguration ozoneConf,
 180   │       List<Pipeline> pipelines) throws IOException {
 181   │     LOG.info("Inside write Pipeline and pipeline size" + pipelines.size());
 182   │     xceiverClients = new ArrayList<>();
 183   │     for (Pipeline p: pipelines){
 184   │       init();
 185   │       LOG.info("run test on pipeline" + p.getId().toString());
 186   │       XceiverClientSpi clientSpi = xceiverClientManager.acquireClient(p);
 187   │       xceiverClients.add(clientSpi);
 188   │       runTest(clientSpi);
 189   │     }
 190   │   }

It has multiple init() calls which reset all the counters. runTest should be called only once IMHO.

[elek]

+1, thanks the update @sadanand48

I have 3 minor/typo comments, but I like this approach

[elek]

Suggested change

	"Coudln't find the any/the selected pipeline");
	"Couldn't find the any/the selected pipeline");

[elek]

Suggested change

LOG.info("Using pipeline {}", firstPipeline.getId());

You don't need this line as you log the same information in the loop bellow.

[elek]

Suggested change

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

[elek]

Merging it. Thanks for the continuous improvement @sadanand48