HADOOP-3577. Tools to inject blocks into name node and simulated data nodes for testing. Contributed by Sanjay Radia.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/trunk@670787 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -7,6 +7,8 @@ Trunk (unreleased changes)
   NEW FEATURES
 
   IMPROVEMENTS
+    HADOOP-3577. Tools to inject blocks into name node and simulated
+    data nodes for testing. (Sanjay Radia via hairong)
 
   OPTIMIZATIONS
 

src/hdfs/org/apache/hadoop/dfs/DataNode.java

@@ -804,6 +804,7 @@ public class DataNode extends Configured
       return false;
     case DatanodeProtocol.DNA_REGISTER:
       // namenode requested a registration - at start or if NN lost contact
+      LOG.info("DatanodeCommand action: DNA_REGISTER");
       register();
       break;
     case DatanodeProtocol.DNA_FINALIZE:
@@ -816,10 +817,14 @@ public class DataNode extends Configured
     case DatanodeProtocol.DNA_BLOCKREPORT:
       // only send BR when receive request the 1st time
       if (waitForFirstBlockReportRequest) {
+        LOG.info("DatanodeCommand action: DNA_BLOCKREPORT - scheduled");
         // dropping all following BR requests
         waitForFirstBlockReportRequest = false;
         // random short delay - helps scatter the BR from all DNs
         scheduleBlockReport(initialBlockReportDelay);
+      } else {
+        LOG.info("DatanodeCommand action: DNA_BLOCKREPORT" +
+            "- ignored becaused one is already scheduled");
       }
       break;
     case DatanodeProtocol.DNA_RECOVERBLOCK:

src/hdfs/org/apache/hadoop/dfs/FSEditLog.java

@@ -733,7 +733,8 @@ class FSEditLog {
     long end = FSNamesystem.now();
     numTransactions++;
     totalTimeTransactions += (end-start);
-    metrics.transactions.inc((end-start));
+    if (metrics != null) // Metrics is non-null only when used inside name node
+      metrics.transactions.inc((end-start));
   }
 
   //
@@ -804,7 +805,8 @@ class FSEditLog {
        this.notifyAll();
     }
 
-    metrics.syncs.inc(elapsed);
+    if (metrics != null) // Metrics is non-null only when used inside name node
+      metrics.syncs.inc(elapsed);
   }
 
   //

src/hdfs/org/apache/hadoop/dfs/Storage.java

@@ -104,7 +104,7 @@ abstract class Storage extends StorageInfo {
   
   private   static final String STORAGE_FILE_LOCK     = "in_use.lock";
   protected static final String STORAGE_FILE_VERSION  = "VERSION";
-  private   static final String STORAGE_DIR_CURRENT   = "current";
+  protected   static final String STORAGE_DIR_CURRENT   = "current";
   private   static final String STORAGE_DIR_PREVIOUS  = "previous";
   private   static final String STORAGE_TMP_REMOVED   = "removed.tmp";
   private   static final String STORAGE_TMP_PREVIOUS  = "previous.tmp";

src/test/org/apache/hadoop/dfs/DataNodeCluster.java

@@ -31,6 +31,8 @@ import org.apache.hadoop.net.DNS;
 
 
 /**
+ * 
+
  * 
  * This program starts a mini cluster of data nodes
  *  (ie a mini cluster without the name node), all within one address space.
@@ -39,35 +41,85 @@ import org.apache.hadoop.net.DNS;
  *  
  *  A use case of this is to run a real name node with a large number of
  *  simulated data nodes for say a NN benchmark.
+ *  
+ * Synopisis:
+ *   DataNodeCluster -n numDatNodes [-r numRacks] -simulated
+ *              [-inject startingBlockId numBlocksPerDN]
+ *
+ * if -simulated is specified then simulated data nodes are started.
+ * if -inject is specified then blocks are injected in each datanode;
+ *    -inject option is valid only for simulated data nodes.
+ *    
+ *    See Also @link #CreateEditsLog for creating a edits log file to
+ *    inject a matching set of blocks into into a name node.
+ *    Typical use of -inject is to inject blocks into a set of datanodes
+ *    using this DataNodeCLuster command
+ *    and then to inject the same blocks into a name node using the
+ *    CreateEditsLog command.
  *
  */
 
 public class DataNodeCluster {
+  static String usage =
+    "Usage: datanodecluster " +
+    " -n <numDataNodes> " + 
+    " [-r <numRacks>] " +
+    " [-simulated] " +
+    " [-inject startingBlockId numBlocksPerDN]" +
+    "\n" + 
+    "  If -r not specified then default rack is used for all Data Nodes\n" +
+    "  Data nodes are simulated if -simulated OR conf file specifies simulated\n";
+  
+  
+  static void printUsageExit() {
+    System.out.println(usage);
+    System.exit(-1); 
+  }
+  static void printUsageExit(String err) {
+    System.out.println(err);
+    printUsageExit();
+  }
   
   public static void main(String[] args) {
     int numDataNodes = 0;
     int numRacks = 0;
+    boolean inject = false;
+    long startingBlockId = 1;
+    int numBlocksPerDNtoInject = 0;
     
     Configuration conf = new Configuration();
-    String usage =
-     "Usage: datanodecluster " +
-     " -n <numDataNodes> " + 
-     " [-r <numRacks>] " +
-     " [-simulated] " +
-     "\n" + 
-     "  If -r not specified then default rack is used for all Data Nodes\n" +
-     "  Data nodes are simulated if -simulated OR conf file specifies simulated\n";
-    
+
     for (int i = 0; i < args.length; i++) { // parse command line
       if (args[i].equals("-n")) {
-        numDataNodes = Integer.parseInt(args[++i]);
+        if (++i >= args.length) {
+          printUsageExit("missing number of nodes" + i + " " + args.length);
+        }
+        numDataNodes = Integer.parseInt(args[i]);
       } else if (args[i].equals("-r")) {
-        numRacks = Integer.parseInt(args[++i]);
+        if (++i >= args.length) {
+          printUsageExit("Missing number of racks");
+        }
+        numRacks = Integer.parseInt(args[i]);
       } else if (args[i].equals("-simulated")) {
         conf.setBoolean(SimulatedFSDataset.CONFIG_PROPERTY_SIMULATED, true);
+      } else if (args[i].equals("-inject")) {
+        if (!conf.getBoolean(SimulatedFSDataset.CONFIG_PROPERTY_SIMULATED,
+                                                                false) ) {
+          System.out.print("-inject is valid only for simulated");
+          printUsageExit(); 
+        }
+       inject = true;
+       if (++i >= args.length) {
+         printUsageExit(
+             "Missing starting block and number of blocks per DN to inject");
+       }
+       startingBlockId = Integer.parseInt(args[i]);
+       if (++i >= args.length) {
+         printUsageExit("Missing number of blocks to inject");
+       }
+       numBlocksPerDNtoInject = Integer.parseInt(args[i]);      
       } else {
-        System.out.println(usage);
-        System.exit(-1);
+        printUsageExit();
       }
     }
     if (numDataNodes <= 0) {
@@ -79,9 +131,13 @@ public class DataNodeCluster {
       System.out.println("No name node address and port in config");
       System.exit(-1);
     }
+    boolean simulated = 
+      conf.getBoolean(SimulatedFSDataset.CONFIG_PROPERTY_SIMULATED, false);
     System.out.println("Starting " + numDataNodes + 
+          (simulated ? "Simulated " : " ") +
           " Data Nodes that will connect to Name Node at " + nameNodeAdr);
   
+    System.setProperty("test.build.data", "/tmp/DataNodeCluster");
 
     MiniDFSCluster mc = new MiniDFSCluster();
     try {
@@ -107,6 +163,23 @@ public class DataNodeCluster {
     try {
       mc.startDataNodes(conf, numDataNodes, true, StartupOption.REGULAR,
           rack4DataNode);
+      if (inject) {
+        long blockSize = 10;
+        System.out.println("Injecting " + numBlocksPerDNtoInject +
+            " blocks in each DN starting at blockId " + startingBlockId +
+            " with blocksize of " + blockSize);
+        Block[] blocks = new Block[numBlocksPerDNtoInject];
+        long blkid = startingBlockId;
+        for (int i_dn = 0; i_dn < numDataNodes; ++i_dn) {
+          for (int i = 0; i < blocks.length; ++i) {
+            blocks[i] = new Block(blkid++, blockSize, 1);
+          }
+          mc.injectBlocks(i_dn, blocks);
+        }
+        System.out.println("Created blocks from Bids " 
+            + startingBlockId + " to "  + (blkid -1));
+      }
+
     } catch (IOException e) {
       System.out.println("Error creating data node:" + e);
     }  

src/test/org/apache/hadoop/dfs/MiniDFSCluster.java

@@ -689,6 +689,9 @@ public class MiniDFSCluster {
    * Wait until the cluster is active and running.
    */
   public void waitActive() throws IOException {
+    if (nameNode == null) {
+      return;
+    }
     InetSocketAddress addr = new InetSocketAddress("localhost",
                                                    getNameNodePort());
     DFSClient client = new DFSClient(addr, conf);

src/test/org/apache/hadoop/dfs/SimulatedFSDataset.java

@@ -250,8 +250,10 @@ public class SimulatedFSDataset  implements FSConstants, FSDatasetInterface, Con
           throw new IOException("Block already exists in  block list");
         }
       }
-      
-      blockMap = new HashMap<Block,BInfo>(injectBlocks.length);
+      HashMap<Block, BInfo> oldBlockMap = blockMap;
+      blockMap = 
+          new HashMap<Block,BInfo>(injectBlocks.length + oldBlockMap.size());
+      blockMap.putAll(oldBlockMap);
       for (Block b: injectBlocks) {
           BInfo binfo = new BInfo(b, false);
           blockMap.put(b, binfo);

src/test/org/apache/hadoop/dfs/TestSimulatedFSDataset.java

@@ -36,7 +36,7 @@ public class TestSimulatedFSDataset extends TestCase {
   
   Configuration conf = null;
   
-  FSDatasetInterface fsdataset = null;
+
   
   static final int NUMBLOCKS = 20;
   static final int BLOCK_LENGTH_MULTIPLIER = 79;
@@ -45,7 +45,7 @@ public class TestSimulatedFSDataset extends TestCase {
     super.setUp();
       conf = new Configuration();
       conf.setBoolean(SimulatedFSDataset.CONFIG_PROPERTY_SIMULATED, true);
-    fsdataset = new SimulatedFSDataset(conf);  
+ 
   }
 
   protected void tearDown() throws Exception {
@@ -56,9 +56,9 @@ public class TestSimulatedFSDataset extends TestCase {
     return blkid*BLOCK_LENGTH_MULTIPLIER;
   }
   
-  int addSomeBlocks() throws IOException {
+  int addSomeBlocks(FSDatasetInterface fsdataset, int startingBlockId) throws IOException {
     int bytesAdded = 0;
-    for (int i = 1; i <= NUMBLOCKS; ++i) {
+    for (int i = startingBlockId; i < startingBlockId+NUMBLOCKS; ++i) {
       Block b = new Block(i, 0, 0); // we pass expected len as zero, - fsdataset should use the sizeof actual data written
       OutputStream dataOut  = fsdataset.writeToBlock(b, false).dataOut;
       assertEquals(0, fsdataset.getLength(b));
@@ -74,8 +74,12 @@ public class TestSimulatedFSDataset extends TestCase {
     }
     return bytesAdded;  
   }
+  int addSomeBlocks(FSDatasetInterface fsdataset ) throws IOException {
+    return addSomeBlocks(fsdataset, 1);
+  }
 
   public void testGetMetaData() throws IOException {
+    FSDatasetInterface fsdataset = new SimulatedFSDataset(conf); 
     Block b = new Block(1, 5, 0);
     try {
       assertFalse(fsdataset.metaFileExists(b));
@@ -83,7 +87,7 @@ public class TestSimulatedFSDataset extends TestCase {
     } catch (IOException e) {
       // ok - as expected
     }
-    addSomeBlocks(); // Only need to add one but ....
+    addSomeBlocks(fsdataset); // Only need to add one but ....
     b = new Block(1, 0, 0);
     InputStream metaInput = fsdataset.getMetaDataInputStream(b);
     DataInputStream metaDataInput = new DataInputStream(metaInput);
@@ -96,9 +100,10 @@ public class TestSimulatedFSDataset extends TestCase {
 
 
   public void testStorageUsage() throws IOException {
+    FSDatasetInterface fsdataset = new SimulatedFSDataset(conf); 
     assertEquals(fsdataset.getDfsUsed(), 0);
     assertEquals(fsdataset.getRemaining(), fsdataset.getCapacity());
-    int bytesAdded = addSomeBlocks();
+    int bytesAdded = addSomeBlocks(fsdataset);
     assertEquals(bytesAdded, fsdataset.getDfsUsed());
     assertEquals(fsdataset.getCapacity()-bytesAdded,  fsdataset.getRemaining());
     
@@ -106,7 +111,8 @@ public class TestSimulatedFSDataset extends TestCase {
 
 
 
-  void  checkBlockDataAndSize(Block b, long expectedLen) throws IOException {
+  void  checkBlockDataAndSize(FSDatasetInterface fsdataset, 
+              Block b, long expectedLen) throws IOException { 
     InputStream input = fsdataset.getBlockInputStream(b);
     long lengthRead = 0;
     int data;
@@ -118,21 +124,35 @@ public class TestSimulatedFSDataset extends TestCase {
   }
   
   public void testWriteRead() throws IOException {
-    addSomeBlocks();
+    FSDatasetInterface fsdataset = new SimulatedFSDataset(conf); 
+    addSomeBlocks(fsdataset);
     for (int i=1; i <= NUMBLOCKS; ++i) {
       Block b = new Block(i, 0, 0);
       assertTrue(fsdataset.isValidBlock(b));
       assertEquals(blockIdToLen(i), fsdataset.getLength(b));
-      checkBlockDataAndSize(b, blockIdToLen(i));
+      checkBlockDataAndSize(fsdataset, b, blockIdToLen(i));
     }
   }
 
 
 
   public void testGetBlockReport() throws IOException {
+    FSDatasetInterface fsdataset = new SimulatedFSDataset(conf); 
+    Block[] blockReport = fsdataset.getBlockReport();
+    assertEquals(0, blockReport.length);
+    int bytesAdded = addSomeBlocks(fsdataset);
+    blockReport = fsdataset.getBlockReport();
+    assertEquals(NUMBLOCKS, blockReport.length);
+    for (Block b: blockReport) {
+      assertNotNull(b);
+      assertEquals(blockIdToLen(b.blkid), b.len);
+    }
+  }
+  public void testInjectionEmpty() throws IOException {
+    FSDatasetInterface fsdataset = new SimulatedFSDataset(conf); 
     Block[] blockReport = fsdataset.getBlockReport();
     assertEquals(0, blockReport.length);
-    int bytesAdded = addSomeBlocks();
+    int bytesAdded = addSomeBlocks(fsdataset);
     blockReport = fsdataset.getBlockReport();
     assertEquals(NUMBLOCKS, blockReport.length);
     for (Block b: blockReport) {
@@ -140,8 +160,8 @@ public class TestSimulatedFSDataset extends TestCase {
       assertEquals(blockIdToLen(b.blkid), b.len);
     }
     
-    // Inject blocks
-    // Now reset fsdataset with an initial block report (Use the blocks we got above)
+    // Inject blocks into an empty fsdataset
+    //  - injecting the blocks we got above.
   
    
     SimulatedFSDataset sfsdataset = new SimulatedFSDataset(conf);
@@ -154,14 +174,51 @@ public class TestSimulatedFSDataset extends TestCase {
       assertEquals(blockIdToLen(b.blkid), sfsdataset.getLength(b));
     }
     assertEquals(bytesAdded, sfsdataset.getDfsUsed());
-    assertEquals(sfsdataset.getCapacity()-bytesAdded,  sfsdataset.getRemaining());
+    assertEquals(sfsdataset.getCapacity()-bytesAdded, sfsdataset.getRemaining());
+  }
 
+  public void testInjectionNonEmpty() throws IOException {
+    FSDatasetInterface fsdataset = new SimulatedFSDataset(conf); 
+    
+    Block[] blockReport = fsdataset.getBlockReport();
+    assertEquals(0, blockReport.length);
+    int bytesAdded = addSomeBlocks(fsdataset);
+    blockReport = fsdataset.getBlockReport();
+    assertEquals(NUMBLOCKS, blockReport.length);
+    for (Block b: blockReport) {
+      assertNotNull(b);
+      assertEquals(blockIdToLen(b.blkid), b.len);
+    }
+    fsdataset = null;
+    
+    // Inject blocks into an non-empty fsdataset
+    //  - injecting the blocks we got above.
+  
+   
+    SimulatedFSDataset sfsdataset = new SimulatedFSDataset(conf);
+    // Add come blocks whose block ids do not conflict with
+    // the ones we are going to inject.
+    bytesAdded += addSomeBlocks(sfsdataset, NUMBLOCKS+1);
+    Block[] blockReport2 = sfsdataset.getBlockReport();
+    assertEquals(NUMBLOCKS, blockReport.length);
+    blockReport2 = sfsdataset.getBlockReport();
+    assertEquals(NUMBLOCKS, blockReport.length);
+    sfsdataset.injectBlocks(blockReport);
+    blockReport = sfsdataset.getBlockReport();
+    assertEquals(NUMBLOCKS*2, blockReport.length);
+    for (Block b: blockReport) {
+      assertNotNull(b);
+      assertEquals(blockIdToLen(b.blkid), b.len);
+      assertEquals(blockIdToLen(b.blkid), sfsdataset.getLength(b));
+    }
+    assertEquals(bytesAdded, sfsdataset.getDfsUsed());
+    assertEquals(sfsdataset.getCapacity()-bytesAdded,  sfsdataset.getRemaining());
+    
     
     // Now test that the dataset cannot be created if it does not have sufficient cap
 
     conf.setLong(SimulatedFSDataset.CONFIG_PROPERTY_CAPACITY, 10);
  
-    
     try {
       sfsdataset = new SimulatedFSDataset(conf);
       sfsdataset.injectBlocks(blockReport);
@@ -172,7 +229,8 @@ public class TestSimulatedFSDataset extends TestCase {
 
   }
 
-  public void checkInvalidBlock(Block b) {
+  public void checkInvalidBlock(Block b) throws IOException {
+    FSDatasetInterface fsdataset = new SimulatedFSDataset(conf); 
     assertFalse(fsdataset.isValidBlock(b));
     try {
       fsdataset.getLength(b);
@@ -198,18 +256,20 @@ public class TestSimulatedFSDataset extends TestCase {
   }
   
   public void testInValidBlocks() throws IOException {
+    FSDatasetInterface fsdataset = new SimulatedFSDataset(conf); 
     Block b = new Block(1, 5, 0);
     checkInvalidBlock(b);
     
     // Now check invlaid after adding some blocks
-    addSomeBlocks();
+    addSomeBlocks(fsdataset);
     b = new Block(NUMBLOCKS + 99, 5, 0);
     checkInvalidBlock(b);
     
   }
 
   public void testInvalidate() throws IOException {
-    int bytesAdded = addSomeBlocks();
+    FSDatasetInterface fsdataset = new SimulatedFSDataset(conf); 
+    int bytesAdded = addSomeBlocks(fsdataset);
     Block[] deleteBlocks = new Block[2];
     deleteBlocks[0] = new Block(1, 0, 0);
     deleteBlocks[1] = new Block(2, 0, 0);