currentValue,
+ long maxValue) {
this.stream = stream;
+ this.currentValue = currentValue;
this.maxValue = maxValue;
}
@@ -366,9 +425,9 @@ private class ProgressBar implements Runnable {
public void run() {
try {
stream.println();
- long keys;
- while((keys = numberOfKeysAdded.get()) < maxValue) {
- print(keys);
+ long value;
+ while((value = currentValue.get()) < maxValue) {
+ print(value);
if(completed) {
break;
}
@@ -389,11 +448,11 @@ public void run() {
/**
* Given current value prints the progress bar.
*
- * @param currentValue
+ * @param value
*/
- private void print(long currentValue) {
+ private void print(long value) {
stream.print('\r');
- double percent = 100.0 * currentValue / maxValue;
+ double percent = 100.0 * value / maxValue;
StringBuilder sb = new StringBuilder();
sb.append(" " + String.format("%.2f", percent) + "% |");
@@ -404,7 +463,7 @@ private void print(long currentValue) {
sb.append(' ');
}
sb.append("| ");
- sb.append(currentValue + "/" + maxValue);
+ sb.append(value + "/" + maxValue);
long timeInSec = TimeUnit.SECONDS.convert(
System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
String timeToPrint = String.format("%d:%02d:%02d", timeInSec / 3600,
@@ -494,10 +553,158 @@ private void printStats(PrintStream out) {
out.println("Time spent in key creation: " + prettyKeyCreationTime);
out.println("Time spent in writing keys: " + prettyKeyWriteTime);
out.println("Total bytes written: " + totalBytesWritten);
+ if(validateWrites) {
+ out.println("Total number of writes validated: " +
+ totalWritesValidated);
+ out.println("Writes validated: " +
+ (100.0 * totalWritesValidated / numberOfKeysAdded.get())
+ + " %");
+ out.println("Successful validation: " +
+ writeValidationSuccessCount);
+ out.println("Unsuccessful validation: " +
+ writeValidationFailureCount);
+ }
out.println("Total Execution time: " + execTime);
out.println("***************************************************");
}
+ /**
+ * Returns the number of volumes created.
+ * @return volume count.
+ */
+ @VisibleForTesting
+ int getNumberOfVolumesCreated() {
+ return numberOfVolumesCreated.get();
+ }
+
+ /**
+ * Returns the number of buckets created.
+ * @return bucket count.
+ */
+ @VisibleForTesting
+ int getNumberOfBucketsCreated() {
+ return numberOfBucketsCreated.get();
+ }
+
+ /**
+ * Returns the number of keys added.
+ * @return keys count.
+ */
+ @VisibleForTesting
+ long getNumberOfKeysAdded() {
+ return numberOfKeysAdded.get();
+ }
+
+ /**
+ * Returns true if random validation of write is enabled.
+ * @return validateWrites
+ */
+ @VisibleForTesting
+ boolean getValidateWrites() {
+ return validateWrites;
+ }
+
+ /**
+ * Returns the number of keys validated.
+ * @return validated key count.
+ */
+ @VisibleForTesting
+ long getTotalKeysValidated() {
+ return totalWritesValidated;
+ }
+
+ /**
+ * Returns the number of successful validation.
+ * @return successful validation count.
+ */
+ @VisibleForTesting
+ long getSuccessfulValidationCount() {
+ return writeValidationSuccessCount;
+ }
+
+ /**
+ * Returns the number of unsuccessful validation.
+ * @return unsuccessful validation count.
+ */
+ @VisibleForTesting
+ long getUnsuccessfulValidationCount() {
+ return writeValidationFailureCount;
+ }
+
+ /**
+ * Validates the write done in ozone cluster.
+ */
+ private class Validator implements Runnable {
+
+ @Override
+ public void run() {
+ while(!completed) {
+ try {
+ KeyValue kv = validationQueue.poll(5, TimeUnit.SECONDS);
+ if(kv != null) {
+ OzoneInputStream is = ozoneClient.
+ getKey(kv.volume, kv.bucket, kv.key);
+ byte[] value = new byte[kv.value.length];
+ int length = is.read(value);
+ totalWritesValidated++;
+ if (length == kv.value.length && Arrays.equals(value, kv.value)) {
+ writeValidationSuccessCount++;
+ } else {
+ writeValidationFailureCount++;
+ LOG.warn("Data validation error for key {}/{}/{}",
+ kv.volume, kv.bucket, kv.key);
+ LOG.warn("Expected: {}, Actual: {}",
+ DFSUtil.bytes2String(kv.value),
+ DFSUtil.bytes2String(value));
+ }
+ }
+ } catch (IOException | InterruptedException ex) {
+ LOG.error("Exception while validating write: " + ex.getMessage());
+ }
+ }
+ }
+ }
+
+
+
+ /**
+ * Wrapper to hold ozone key-value pair.
+ */
+ private static class KeyValue {
+
+ /**
+ * Volume name associated with the key-value.
+ */
+ private String volume;
+
+ /**
+ * Bucket name associated with the key-value.
+ */
+ private String bucket;
+ /**
+ * Key name associated with the key-value.
+ */
+ private String key;
+ /**
+ * Value associated with the key-value.
+ */
+ private byte[] value;
+
+ /**
+ * Constructs a new ozone key-value pair.
+ *
+ * @param key key part
+ * @param value value part
+ */
+ KeyValue(
+ String volume, String bucket, String key, byte[] value) {
+ this.volume = volume;
+ this.bucket = bucket;
+ this.key = key;
+ this.value = value;
+ }
+ }
+
/**
* @param args arguments
*/
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/tools/TestCorona.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/tools/TestCorona.java
new file mode 100644
index 00000000000..7e87b86322f
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/tools/TestCorona.java
@@ -0,0 +1,117 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.tools;
+
+import org.apache.hadoop.ozone.MiniOzoneCluster;
+import org.apache.hadoop.ozone.OzoneConfigKeys;
+import org.apache.hadoop.ozone.OzoneConfiguration;
+import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.util.ToolRunner;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Tests Corona, with MiniOzoneCluster.
+ */
+public class TestCorona {
+
+ private static MiniOzoneCluster cluster;
+ private static OzoneConfiguration conf;
+
+ /**
+ * Create a MiniDFSCluster for testing.
+ *
+ * Ozone is made active by setting OZONE_ENABLED = true and
+ * OZONE_HANDLER_TYPE_KEY = "distributed"
+ *
+ * @throws IOException
+ */
+ @BeforeClass
+ public static void init() throws Exception {
+ conf = new OzoneConfiguration();
+ conf.set(OzoneConfigKeys.OZONE_HANDLER_TYPE_KEY,
+ OzoneConsts.OZONE_HANDLER_DISTRIBUTED);
+ cluster = new MiniOzoneCluster.Builder(conf)
+ .setHandlerType(OzoneConsts.OZONE_HANDLER_DISTRIBUTED).build();
+ }
+
+ /**
+ * Shutdown MiniDFSCluster.
+ */
+ @AfterClass
+ public static void shutdown() {
+ if (cluster != null) {
+ cluster.shutdown();
+ }
+ }
+
+ @Test
+ public void defaultTest() throws Exception {
+ List args = new ArrayList<>();
+ args.add("-numOfVolumes");
+ args.add("2");
+ args.add("-numOfBuckets");
+ args.add("5");
+ args.add("-numOfKeys");
+ args.add("10");
+ Corona corona = new Corona(conf);
+ int res = ToolRunner.run(conf, corona,
+ args.toArray(new String[0]));
+ Assert.assertEquals(2, corona.getNumberOfVolumesCreated());
+ Assert.assertEquals(10, corona.getNumberOfBucketsCreated());
+ Assert.assertEquals(100, corona.getNumberOfKeysAdded());
+ Assert.assertEquals(0, res);
+ }
+
+ @Test
+ public void validateWriteTest() throws Exception {
+ PrintStream originalStream = System.out;
+ ByteArrayOutputStream outStream = new ByteArrayOutputStream();
+ System.setOut(new PrintStream(outStream));
+ List args = new ArrayList<>();
+ args.add("-validateWrites");
+ args.add("-numOfVolumes");
+ args.add("2");
+ args.add("-numOfBuckets");
+ args.add("5");
+ args.add("-numOfKeys");
+ args.add("10");
+ Corona corona = new Corona(conf);
+ int res = ToolRunner.run(conf, corona,
+ args.toArray(new String[0]));
+ Assert.assertEquals(0, res);
+ Assert.assertEquals(2, corona.getNumberOfVolumesCreated());
+ Assert.assertEquals(10, corona.getNumberOfBucketsCreated());
+ Assert.assertEquals(100, corona.getNumberOfKeysAdded());
+ Assert.assertTrue(corona.getValidateWrites());
+ Assert.assertNotEquals(0, corona.getTotalKeysValidated());
+ Assert.assertNotEquals(0, corona.getSuccessfulValidationCount());
+ Assert.assertEquals(0, corona.getUnsuccessfulValidationCount());
+ System.setOut(originalStream);
+ }
+
+}
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/tools/package-info.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/tools/package-info.java
new file mode 100644
index 00000000000..ea56345352e
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/tools/package-info.java
@@ -0,0 +1,21 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.ozone.tools;
+/**
+ * Classes related to Ozone tools tests.
+ */