FSImage implements Closeable { public synchronized boolean saveNamespace(long timeWindow, long txGap, FSNamesystem source) throws IOException { if (timeWindow > 0 || txGap > 0) { final FSImageStorageInspector inspector = storage.readAndInspectDirs( EnumSet.of(NameNodeFile.IMAGE, NameNodeFile.IMAGE_ROLLBACK), StartupOption.REGULAR); FSImageFile image = inspector.getLatestImages().get(0); File imageFile = image.getFile(); final long checkpointTxId = image.getCheckpointTxId(); final long checkpointAge = Time.now() - imageFile.lastModified(); if (checkpointAge <= timeWindow * 1000 && checkpointTxId >= this.getCorrectLastAppliedOrWrittenTxId() - txGap) { return false; } } saveNamespace(source, NameNodeFile.IMAGE, null); return true; } FSImage(Configuration conf); protected  FSImage(Configuration conf,
                    Collection<URI> imageDirs,
                    List<URI> editsDirs); static boolean recoverStorageDirs(StartupOption startOpt,
      NNStorage storage, Map<StorageDirectory, StorageState> dataDirStates); static void checkUpgrade(NNStorage storage); boolean hasRollbackFSImage(); FSEditLog getEditLog(); void initEditLog(StartupOption startOpt); long loadEdits(Iterable<EditLogInputStream> editStreams,
      FSNamesystem target); long loadEdits(Iterable<EditLogInputStream> editStreams,
      FSNamesystem target, long maxTxnsToRead,
      StartupOption startOpt, MetaRecoveryContext recovery); void saveLegacyOIVImage(FSNamesystem source, String targetDir,
      Canceler canceler); synchronized void updateStorageVersion(); synchronized boolean saveNamespace(long timeWindow, long txGap,
      FSNamesystem source); void saveNamespace(FSNamesystem source); synchronized void saveNamespace(FSNamesystem source, NameNodeFile nnf,
      Canceler canceler); boolean addToCheckpointing(long txid); void removeFromCheckpointing(long txid); synchronized void saveDigestAndRenameCheckpointImage(NameNodeFile nnf,
      long txid, MD5Hash digest); @Override synchronized void close(); NNStorage getStorage(); int getLayoutVersion(); int getNamespaceID(); String getClusterID(); String getBlockPoolID(); synchronized long getLastAppliedTxId(); long getLastAppliedOrWrittenTxId(); long getCorrectLastAppliedOrWrittenTxId(); void updateLastAppliedTxIdFromWritten(); long getMostRecentCheckpointTxId(); static final Logger LOG; static final int SHUTDOWN_HOOK_PRIORITY; }
@Test public void testDigest() throws IOException { Configuration conf = new Configuration(); MiniDFSCluster cluster = null; try { cluster = new MiniDFSCluster.Builder(conf).numDataNodes(0).build(); DistributedFileSystem fs = cluster.getFileSystem(); fs.setSafeMode(SafeModeAction.SAFEMODE_ENTER); fs.saveNamespace(); fs.setSafeMode(SafeModeAction.SAFEMODE_LEAVE); File currentDir = FSImageTestUtil.getNameNodeCurrentDirs(cluster, 0).get( 0); File fsimage = FSImageTestUtil.findNewestImageFile(currentDir .getAbsolutePath()); assertEquals(MD5FileUtils.readStoredMd5ForFile(fsimage), MD5FileUtils.computeMd5ForFile(fsimage)); } finally { if (cluster != null) { cluster.shutdown(); } } }
@Test(timeout = 60000) public void testSupportBlockGroup() throws Exception { final short GROUP_SIZE = (short) (testECPolicy.getNumDataUnits() + testECPolicy.getNumParityUnits()); final int BLOCK_SIZE = 8 * 1024 * 1024; Configuration conf = new HdfsConfiguration(); conf.setLong(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, BLOCK_SIZE); MiniDFSCluster cluster = null; try { cluster = new MiniDFSCluster.Builder(conf).numDataNodes(GROUP_SIZE) .build(); cluster.waitActive(); DistributedFileSystem fs = cluster.getFileSystem(); DFSTestUtil.enableAllECPolicies(fs); Path parentDir = new Path("/ec-10-4"); Path childDir = new Path(parentDir, "ec-3-2"); ErasureCodingPolicy ec32Policy = SystemErasureCodingPolicies .getByID(SystemErasureCodingPolicies.RS_3_2_POLICY_ID); fs.mkdirs(parentDir); fs.mkdirs(childDir); fs.setErasureCodingPolicy(parentDir, testECPolicy.getName()); fs.setErasureCodingPolicy(childDir, ec32Policy.getName()); Path file_10_4 = new Path(parentDir, "striped_file_10_4"); Path file_3_2 = new Path(childDir, "striped_file_3_2"); byte[] bytes = StripedFileTestUtil.generateBytes(BLOCK_SIZE); DFSTestUtil.writeFile(fs, file_10_4, new String(bytes)); DFSTestUtil.writeFile(fs, file_3_2, new String(bytes)); fs.setSafeMode(SafeModeAction.SAFEMODE_ENTER); fs.saveNamespace(); fs.setSafeMode(SafeModeAction.SAFEMODE_LEAVE); cluster.restartNameNodes(); fs = cluster.getFileSystem(); assertTrue(fs.exists(file_10_4)); assertTrue(fs.exists(file_3_2)); FSNamesystem fsn = cluster.getNamesystem(); INodeFile inode = fsn.dir.getINode(file_10_4.toString()).asFile(); assertTrue(inode.isStriped()); assertEquals(testECPolicy.getId(), inode.getErasureCodingPolicyID()); BlockInfo[] blks = inode.getBlocks(); assertEquals(1, blks.length); assertTrue(blks[0].isStriped()); assertEquals(testECPolicy.getId(), fs.getErasureCodingPolicy(file_10_4).getId()); assertEquals(testECPolicy.getId(), ((BlockInfoStriped)blks[0]).getErasureCodingPolicy().getId()); assertEquals(testECPolicy.getNumDataUnits(), ((BlockInfoStriped) blks[0]).getDataBlockNum()); assertEquals(testECPolicy.getNumParityUnits(), ((BlockInfoStriped) blks[0]).getParityBlockNum()); byte[] content = DFSTestUtil.readFileAsBytes(fs, file_10_4); assertArrayEquals(bytes, content); inode = fsn.dir.getINode(file_3_2.toString()).asFile(); assertTrue(inode.isStriped()); assertEquals(SystemErasureCodingPolicies.getByID( SystemErasureCodingPolicies.RS_3_2_POLICY_ID).getId(), inode.getErasureCodingPolicyID()); blks = inode.getBlocks(); assertEquals(1, blks.length); assertTrue(blks[0].isStriped()); assertEquals(ec32Policy.getId(), fs.getErasureCodingPolicy(file_3_2).getId()); assertEquals(ec32Policy.getNumDataUnits(), ((BlockInfoStriped) blks[0]).getDataBlockNum()); assertEquals(ec32Policy.getNumParityUnits(), ((BlockInfoStriped) blks[0]).getParityBlockNum()); content = DFSTestUtil.readFileAsBytes(fs, file_3_2); assertArrayEquals(bytes, content); ErasureCodingPolicy ecPolicy = fsn.getErasureCodingPolicy(parentDir.toString()); assertNotNull(ecPolicy); assertEquals(testECPolicy.getId(), ecPolicy.getId()); ecPolicy = fsn.getErasureCodingPolicy(childDir.toString()); assertNotNull(ecPolicy); assertEquals(ec32Policy.getId(), ecPolicy.getId()); ecPolicy = fsn.getErasureCodingPolicy("/"); assertNull(ecPolicy); } finally { if (cluster != null) { cluster.shutdown(); } } }
@Test public void testHasNonEcBlockUsingStripedIDForLoadFile() throws IOException{ Configuration conf = new HdfsConfiguration(); MiniDFSCluster cluster = null; try { cluster = new MiniDFSCluster.Builder(conf).numDataNodes(9) .build(); cluster.waitActive(); DistributedFileSystem fs = cluster.getFileSystem(); FSNamesystem fns = cluster.getNamesystem(); String testDir = "/test_block_manager"; String testFile = "testfile_loadfile"; String testFilePath = testDir + "/" + testFile; String clientName = "testUser_loadfile"; String clientMachine = "testMachine_loadfile"; long blkId = -1; long blkNumBytes = 1024; long timestamp = 1426222918; fs.mkdir(new Path(testDir), new FsPermission("755")); Path p = new Path(testFilePath); DFSTestUtil.createFile(fs, p, 0, (short) 1, 1); BlockInfoContiguous cBlk = new BlockInfoContiguous( new Block(blkId, blkNumBytes, timestamp), (short)3); INodeFile file = (INodeFile)fns.getFSDirectory().getINode(testFilePath); file.toUnderConstruction(clientName, clientMachine); file.addBlock(cBlk); TestINodeFile.toCompleteFile(file); fns.enterSafeMode(false); fns.saveNamespace(0, 0); cluster.restartNameNodes(); cluster.waitActive(); fns = cluster.getNamesystem(); assertTrue(fns.getBlockManager().hasNonEcBlockUsingStripedID()); fs = cluster.getFileSystem(); fs.delete(p,false); fns.enterSafeMode(false); fns.saveNamespace(0, 0); cluster.restartNameNodes(); cluster.waitActive(); fns = cluster.getNamesystem(); assertFalse(fns.getBlockManager().hasNonEcBlockUsingStripedID()); cluster.shutdown(); cluster = null; } finally { if (cluster != null) { cluster.shutdown(); } } }
@Test public void testHasNonEcBlockUsingStripedIDForLoadUCFile() throws IOException{ Configuration conf = new HdfsConfiguration(); MiniDFSCluster cluster = null; try { cluster = new MiniDFSCluster.Builder(conf).numDataNodes(9) .build(); cluster.waitActive(); DistributedFileSystem fs = cluster.getFileSystem(); FSNamesystem fns = cluster.getNamesystem(); String testDir = "/test_block_manager"; String testFile = "testfile_loaducfile"; String testFilePath = testDir + "/" + testFile; String clientName = "testUser_loaducfile"; String clientMachine = "testMachine_loaducfile"; long blkId = -1; long blkNumBytes = 1024; long timestamp = 1426222918; fs.mkdir(new Path(testDir), new FsPermission("755")); Path p = new Path(testFilePath); DFSTestUtil.createFile(fs, p, 0, (short) 1, 1); BlockInfoContiguous cBlk = new BlockInfoContiguous( new Block(blkId, blkNumBytes, timestamp), (short)3); INodeFile file = (INodeFile)fns.getFSDirectory().getINode(testFilePath); file.toUnderConstruction(clientName, clientMachine); file.addBlock(cBlk); fns.enterSafeMode(false); fns.saveNamespace(0, 0); cluster.restartNameNodes(); cluster.waitActive(); fns = cluster.getNamesystem(); assertTrue(fns.getBlockManager().hasNonEcBlockUsingStripedID()); cluster.shutdown(); cluster = null; } finally { if (cluster != null) { cluster.shutdown(); } } }
@Test public void testHasNonEcBlockUsingStripedIDForLoadSnapshot() throws IOException{ Configuration conf = new HdfsConfiguration(); MiniDFSCluster cluster = null; try { cluster = new MiniDFSCluster.Builder(conf).numDataNodes(9) .build(); cluster.waitActive(); DistributedFileSystem fs = cluster.getFileSystem(); FSNamesystem fns = cluster.getNamesystem(); String testDir = "/test_block_manager"; String testFile = "testfile_loadSnapshot"; String testFilePath = testDir + "/" + testFile; String clientName = "testUser_loadSnapshot"; String clientMachine = "testMachine_loadSnapshot"; long blkId = -1; long blkNumBytes = 1024; long timestamp = 1426222918; Path d = new Path(testDir); fs.mkdir(d, new FsPermission("755")); fs.allowSnapshot(d); Path p = new Path(testFilePath); DFSTestUtil.createFile(fs, p, 0, (short) 1, 1); BlockInfoContiguous cBlk = new BlockInfoContiguous( new Block(blkId, blkNumBytes, timestamp), (short)3); INodeFile file = (INodeFile)fns.getFSDirectory().getINode(testFilePath); file.toUnderConstruction(clientName, clientMachine); file.addBlock(cBlk); TestINodeFile.toCompleteFile(file); fs.createSnapshot(d,"testHasNonEcBlockUsingStripeID"); fs.truncate(p,0); fns.enterSafeMode(false); fns.saveNamespace(0, 0); cluster.restartNameNodes(); cluster.waitActive(); fns = cluster.getNamesystem(); assertTrue(fns.getBlockManager().hasNonEcBlockUsingStripedID()); cluster.shutdown(); cluster = null; } finally { if (cluster != null) { cluster.shutdown(); } } }
@Test public void testSaveAndLoadErasureCodingPolicies() throws IOException{ Configuration conf = new Configuration(); final int blockSize = 16 * 1024 * 1024; conf.setLong(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, blockSize); try (MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf).numDataNodes(10).build()) { cluster.waitActive(); DistributedFileSystem fs = cluster.getFileSystem(); DFSTestUtil.enableAllECPolicies(fs); fs.setSafeMode(SafeModeAction.SAFEMODE_ENTER); fs.saveNamespace(); fs.setSafeMode(SafeModeAction.SAFEMODE_LEAVE); cluster.restartNameNodes(); cluster.waitActive(); assertEquals("Erasure coding policy number should match", SystemErasureCodingPolicies.getPolicies().size(), ErasureCodingPolicyManager.getInstance().getPolicies().length); ECSchema newSchema = new ECSchema("rs", 5, 4); ErasureCodingPolicy newPolicy = new ErasureCodingPolicy(newSchema, 2 * 1024, (byte) 254); ErasureCodingPolicy[] policies = new ErasureCodingPolicy[]{newPolicy}; AddErasureCodingPolicyResponse[] ret = fs.addErasureCodingPolicies(policies); assertEquals(1, ret.length); assertEquals(true, ret[0].isSucceed()); newPolicy = ret[0].getPolicy(); fs.setSafeMode(SafeModeAction.SAFEMODE_ENTER); fs.saveNamespace(); fs.setSafeMode(SafeModeAction.SAFEMODE_LEAVE); cluster.restartNameNodes(); cluster.waitActive(); assertEquals("Erasure coding policy number should match", SystemErasureCodingPolicies.getPolicies().size() + 1, ErasureCodingPolicyManager.getInstance().getPolicies().length); ErasureCodingPolicy ecPolicy = ErasureCodingPolicyManager.getInstance().getByID(newPolicy.getId()); assertEquals("Newly added erasure coding policy is not found", newPolicy, ecPolicy); assertEquals( "Newly added erasure coding policy should be of disabled state", ErasureCodingPolicyState.DISABLED, DFSTestUtil.getECPolicyState(ecPolicy)); testChangeErasureCodingPolicyState(cluster, blockSize, newPolicy, false); testChangeErasureCodingPolicyState(cluster, blockSize, SystemErasureCodingPolicies.getByID((byte) 1), true); testChangeErasureCodingPolicyState(cluster, blockSize, SystemErasureCodingPolicies.getByID((byte) 2), false); } }