Mover { private ExitStatus run() { try { init(); return new Processor().processNamespace().getExitStatus(); } catch (IllegalArgumentException e) { System.out.println(e + ". Exiting ..."); return ExitStatus.ILLEGAL_ARGUMENTS; } catch (IOException e) { System.out.println(e + ". Exiting ..."); LOG.error(e + ". Exiting ..."); return ExitStatus.IO_EXCEPTION; } finally { dispatcher.shutdownNow(); } } Mover(NameNodeConnector nnc, Configuration conf, AtomicInteger retryCount,
      Map<Long, Set<DatanodeInfo>> excludedPinnedBlocks); static void main(String[] args);  }
@Test(timeout = 300000) public void testWithFederateClusterWithinSameNode() throws Exception { final Configuration conf = new HdfsConfiguration(); initConf(conf); final MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf) .numDataNodes(4).storageTypes( new StorageType[] {StorageType.DISK, StorageType.ARCHIVE}).nnTopology(MiniDFSNNTopology .simpleFederatedTopology(2)).build(); DFSTestUtil.setFederatedConfiguration(cluster, conf); try { cluster.waitActive(); final String file = "/test/file"; Path dir = new Path ("/test"); final DistributedFileSystem dfs1 = cluster.getFileSystem(0); final DistributedFileSystem dfs2 = cluster.getFileSystem(1); URI nn1 = dfs1.getUri(); URI nn2 = dfs2.getUri(); setupStoragePoliciesAndPaths(dfs1, dfs2, dir, file); dfs1.setStoragePolicy(dir, "COLD"); int rc = ToolRunner.run(conf, new Mover.Cli(), new String[] {"-p", nn1 + dir.toString()}); Assert.assertEquals("Movement to ARCHIVE should be successful", 0, rc); dfs2.setStoragePolicy(dir, "HOT"); rc = ToolRunner.run(conf, new Mover.Cli(), new String[] {"-p", nn2 + dir.toString()}); Assert.assertEquals("Movement to DISK should be successful", 0, rc); waitForLocatedBlockWithArchiveStorageType(dfs1, file, 3); waitForLocatedBlockWithDiskStorageType(dfs2, file, 3); } finally { cluster.shutdown(); } }
@Test(timeout = 300000) public void testWithFederatedCluster() throws Exception{ final Configuration conf = new HdfsConfiguration(); initConf(conf); final MiniDFSCluster cluster = new MiniDFSCluster .Builder(conf) .storageTypes(new StorageType[]{StorageType.DISK, StorageType.ARCHIVE}) .nnTopology(MiniDFSNNTopology.simpleFederatedTopology(2)) .numDataNodes(4).build(); DFSTestUtil.setFederatedConfiguration(cluster, conf); try { cluster.waitActive(); final String file = "/test/file"; Path dir = new Path ("/test"); final DistributedFileSystem dfs1 = cluster.getFileSystem(0); final DistributedFileSystem dfs2 = cluster.getFileSystem(1); URI nn1 = dfs1.getUri(); URI nn2 = dfs2.getUri(); setupStoragePoliciesAndPaths(dfs1, dfs2, dir, file); dfs1.setStoragePolicy(dir, "COLD"); dfs2.setStoragePolicy(dir, "HOT"); int rc = ToolRunner.run(conf, new Mover.Cli(), new String[] {"-p", nn1 + dir.toString(), nn2 + dir.toString()}); Assert.assertEquals("Movement to DISK should be successful", 0, rc); waitForLocatedBlockWithArchiveStorageType(dfs1, file, 3); waitForLocatedBlockWithDiskStorageType(dfs2, file, 3); } finally { cluster.shutdown(); } }
@Test(timeout = 300000) public void testWithFederatedHACluster() throws Exception{ final Configuration conf = new HdfsConfiguration(); initConf(conf); final MiniDFSCluster cluster = new MiniDFSCluster .Builder(conf) .storageTypes(new StorageType[]{StorageType.DISK, StorageType.ARCHIVE}) .nnTopology(MiniDFSNNTopology.simpleHAFederatedTopology(2)) .numDataNodes(4).build(); DFSTestUtil.setFederatedHAConfiguration(cluster, conf); try { Collection<URI> namenodes = DFSUtil.getInternalNsRpcUris(conf); Iterator<URI> iter = namenodes.iterator(); URI nn1 = iter.next(); URI nn2 = iter.next(); cluster.transitionToActive(0); cluster.transitionToActive(2); final String file = "/test/file"; Path dir = new Path ("/test"); final DistributedFileSystem dfs1 = (DistributedFileSystem) FileSystem .get(nn1, conf); final DistributedFileSystem dfs2 = (DistributedFileSystem) FileSystem .get(nn2, conf); setupStoragePoliciesAndPaths(dfs1, dfs2, dir, file); dfs1.setStoragePolicy(dir, "COLD"); dfs2.setStoragePolicy(dir, "HOT"); int rc = ToolRunner.run(conf, new Mover.Cli(), new String[] {"-p", nn1 + dir.toString(), nn2 + dir.toString()}); Assert.assertEquals("Movement to DISK should be successful", 0, rc); waitForLocatedBlockWithArchiveStorageType(dfs1, file, 3); waitForLocatedBlockWithDiskStorageType(dfs2, file, 3); } finally { cluster.shutdown(); } }
@Test(timeout = 300000) public void testTwoReplicaSameStorageTypeShouldNotSelect() throws Exception { final Configuration conf = new HdfsConfiguration(); initConf(conf); final MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf) .numDataNodes(3) .storageTypes( new StorageType[][] { { StorageType.DISK, StorageType.ARCHIVE }, { StorageType.DISK, StorageType.DISK }, { StorageType.DISK, StorageType.ARCHIVE } }).build(); try { cluster.waitActive(); final DistributedFileSystem dfs = cluster.getFileSystem(); final String file = "/testForTwoReplicaSameStorageTypeShouldNotSelect"; final FSDataOutputStream out = dfs.create(new Path(file), (short) 2); out.writeChars("testForTwoReplicaSameStorageTypeShouldNotSelect"); out.close(); LocatedBlock lb = dfs.getClient().getLocatedBlocks(file, 0).get(0); StorageType[] storageTypes = lb.getStorageTypes(); for (StorageType storageType : storageTypes) { Assert.assertTrue(StorageType.DISK == storageType); } dfs.setStoragePolicy(new Path(file), "COLD"); int rc = ToolRunner.run(conf, new Mover.Cli(), new String[] { "-p", file.toString() }); Assert.assertEquals("Movement to ARCHIVE should be successful", 0, rc); waitForLocatedBlockWithArchiveStorageType(dfs, file, 2); } finally { cluster.shutdown(); } }
@Test(timeout = 300000) public void testMoveWhenStoragePolicyNotSatisfying() throws Exception { final Configuration conf = new HdfsConfiguration(); conf.set(DFSConfigKeys.DFS_STORAGE_POLICY_SATISFIER_MODE_KEY, StoragePolicySatisfierMode.NONE.toString()); final MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf) .numDataNodes(3) .storageTypes( new StorageType[][] { { StorageType.DISK }, { StorageType.DISK }, { StorageType.DISK } }).build(); try { cluster.waitActive(); final DistributedFileSystem dfs = cluster.getFileSystem(); final String file = "/testMoveWhenStoragePolicyNotSatisfying"; final FSDataOutputStream out = dfs.create(new Path(file)); out.writeChars("testMoveWhenStoragePolicyNotSatisfying"); out.close(); dfs.setStoragePolicy(new Path(file), "COLD"); int rc = ToolRunner.run(conf, new Mover.Cli(), new String[] { "-p", file.toString() }); int exitcode = ExitStatus.NO_MOVE_BLOCK.getExitCode(); Assert.assertEquals("Exit code should be " + exitcode, exitcode, rc); } finally { cluster.shutdown(); } }
@Test(timeout = 300000) public void testMoveWhenStoragePolicySatisfierIsRunning() throws Exception { final Configuration conf = new HdfsConfiguration(); conf.set(DFSConfigKeys.DFS_STORAGE_POLICY_SATISFIER_MODE_KEY, StoragePolicySatisfierMode.EXTERNAL.toString()); final MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf) .numDataNodes(3) .storageTypes( new StorageType[][] {{StorageType.DISK}, {StorageType.DISK}, {StorageType.DISK}}).build(); try { cluster.waitActive(); DFSTestUtil.getNameNodeConnector(conf, HdfsServerConstants.MOVER_ID_PATH, 1, true); final DistributedFileSystem dfs = cluster.getFileSystem(); final String file = "/testMoveWhenStoragePolicySatisfierIsRunning"; final FSDataOutputStream out = dfs.create(new Path(file)); out.writeChars("testMoveWhenStoragePolicySatisfierIsRunning"); out.close(); dfs.setStoragePolicy(new Path(file), "COLD"); int rc = ToolRunner.run(conf, new Mover.Cli(), new String[] {"-p", file.toString()}); int exitcode = ExitStatus.IO_EXCEPTION.getExitCode(); Assert.assertEquals("Exit code should be " + exitcode, exitcode, rc); } finally { cluster.shutdown(); } }
@Test public void testMoverFailedRetry() throws Exception { final Configuration conf = new HdfsConfiguration(); initConf(conf); conf.set(DFSConfigKeys.DFS_MOVER_RETRY_MAX_ATTEMPTS_KEY, "2"); final MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf) .numDataNodes(3) .storageTypes( new StorageType[][] {{StorageType.DISK, StorageType.ARCHIVE}, {StorageType.DISK, StorageType.ARCHIVE}, {StorageType.DISK, StorageType.ARCHIVE}}).build(); try { cluster.waitActive(); final DistributedFileSystem dfs = cluster.getFileSystem(); final String file = "/testMoverFailedRetry"; final FSDataOutputStream out = dfs.create(new Path(file), (short) 2); out.writeChars("testMoverFailedRetry"); out.close(); LocatedBlock lb = dfs.getClient().getLocatedBlocks(file, 0).get(0); cluster.corruptBlockOnDataNodesByDeletingBlockFile(lb.getBlock()); dfs.setStoragePolicy(new Path(file), "COLD"); int rc = ToolRunner.run(conf, new Mover.Cli(), new String[] {"-p", file.toString()}); Assert.assertEquals("Movement should fail after some retry", ExitStatus.NO_MOVE_PROGRESS.getExitCode(), rc); } finally { cluster.shutdown(); } }
@Test(timeout=100000) public void testBalancerMaxIterationTimeNotAffectMover() throws Exception { long blockSize = 10*1024*1024; final Configuration conf = new HdfsConfiguration(); initConf(conf); conf.setInt(DFSConfigKeys.DFS_MOVER_MOVERTHREADS_KEY, 1); conf.setInt( DFSConfigKeys.DFS_DATANODE_BALANCE_MAX_NUM_CONCURRENT_MOVES_KEY, 1); conf.setLong(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, blockSize); conf.setLong(DFSConfigKeys.DFS_BYTES_PER_CHECKSUM_KEY, blockSize); conf.setLong(DFSConfigKeys.DFS_BALANCER_MAX_ITERATION_TIME_KEY, 200L); conf.setInt(DFSConfigKeys.DFS_MOVER_RETRY_MAX_ATTEMPTS_KEY, 1); conf.setLong(DFSConfigKeys.DFS_CLIENT_SOCKET_TIMEOUT_KEY, 1000L); final MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf) .numDataNodes(2) .storageTypes( new StorageType[][] {{StorageType.DISK, StorageType.DISK}, {StorageType.ARCHIVE, StorageType.ARCHIVE}}) .build(); try { cluster.waitActive(); final DistributedFileSystem fs = cluster.getFileSystem(); final String file = "/testMaxIterationTime.dat"; final Path path = new Path(file); short rep_factor = 1; int seed = 0xFAFAFA; DFSTestUtil.createFile(fs, path, 4L * blockSize, rep_factor, seed); fs.setStoragePolicy(new Path(file), "COLD"); int rc = ToolRunner.run(conf, new Mover.Cli(), new String[] {"-p", file}); Assert.assertEquals("Retcode expected to be ExitStatus.SUCCESS (0).", ExitStatus.SUCCESS.getExitCode(), rc); } finally { cluster.shutdown(); } }
@Test(timeout = 300000) public void testMoverWithStripedFile() throws Exception { final Configuration conf = new HdfsConfiguration(); initConfWithStripe(conf); int numOfDatanodes =10; int storagesPerDatanode=2; long capacity = 10 * defaultBlockSize; long[][] capacities = new long[numOfDatanodes][storagesPerDatanode]; for (int i = 0; i < numOfDatanodes; i++) { for(int j=0;j<storagesPerDatanode;j++){ capacities[i][j]=capacity; } } final MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf) .numDataNodes(numOfDatanodes) .storagesPerDatanode(storagesPerDatanode) .storageTypes(new StorageType[][]{ {StorageType.DISK, StorageType.DISK}, {StorageType.DISK, StorageType.DISK}, {StorageType.DISK, StorageType.DISK}, {StorageType.DISK, StorageType.DISK}, {StorageType.DISK, StorageType.DISK}, {StorageType.DISK, StorageType.ARCHIVE}, {StorageType.DISK, StorageType.ARCHIVE}, {StorageType.DISK, StorageType.ARCHIVE}, {StorageType.DISK, StorageType.ARCHIVE}, {StorageType.DISK, StorageType.ARCHIVE}}) .storageCapacities(capacities) .build(); try { cluster.waitActive(); cluster.getFileSystem().enableErasureCodingPolicy( StripedFileTestUtil.getDefaultECPolicy().getName()); ClientProtocol client = NameNodeProxies.createProxy(conf, cluster.getFileSystem(0).getUri(), ClientProtocol.class).getProxy(); String barDir = "/bar"; client.mkdirs(barDir, new FsPermission((short) 777), true); client.setStoragePolicy(barDir, HdfsConstants.HOT_STORAGE_POLICY_NAME); client.setErasureCodingPolicy(barDir, StripedFileTestUtil.getDefaultECPolicy().getName()); final String fooFile = "/bar/foo"; long fileLen = 20 * defaultBlockSize; DFSTestUtil.createFile(cluster.getFileSystem(), new Path(fooFile), fileLen,(short) 3, 0); LocatedBlocks locatedBlocks = client.getBlockLocations(fooFile, 0, fileLen); for(LocatedBlock lb : locatedBlocks.getLocatedBlocks()){ for( StorageType type : lb.getStorageTypes()){ Assert.assertEquals(StorageType.DISK, type); } } StripedFileTestUtil.verifyLocatedStripedBlocks(locatedBlocks, dataBlocks + parityBlocks); numOfDatanodes +=5; capacities = new long[5][storagesPerDatanode]; for (int i = 0; i < 5; i++) { for(int j=0;j<storagesPerDatanode;j++){ capacities[i][j]=capacity; } } cluster.startDataNodes(conf, 5, new StorageType[][]{ {StorageType.ARCHIVE, StorageType.ARCHIVE}, {StorageType.ARCHIVE, StorageType.ARCHIVE}, {StorageType.ARCHIVE, StorageType.ARCHIVE}, {StorageType.ARCHIVE, StorageType.ARCHIVE}, {StorageType.ARCHIVE, StorageType.ARCHIVE}}, true, null, null, null,capacities, null, false, false, false, null); cluster.triggerHeartbeats(); client.setStoragePolicy(barDir, "COLD"); int rc = ToolRunner.run(conf, new Mover.Cli(), new String[] { "-p", barDir }); Assert.assertEquals("Movement to ARCHIVE should be successful", 0, rc); locatedBlocks = client.getBlockLocations(fooFile, 0, fileLen); for(LocatedBlock lb : locatedBlocks.getLocatedBlocks()){ for( StorageType type : lb.getStorageTypes()){ Assert.assertEquals(StorageType.ARCHIVE, type); } } StripedFileTestUtil.verifyLocatedStripedBlocks(locatedBlocks, dataBlocks + parityBlocks); numOfDatanodes += 5; capacities = new long[5][storagesPerDatanode]; for (int i = 0; i < 5; i++) { for (int j = 0; j < storagesPerDatanode; j++) { capacities[i][j] = capacity; } } cluster.startDataNodes(conf, 5, new StorageType[][] { { StorageType.SSD, StorageType.DISK }, { StorageType.SSD, StorageType.DISK }, { StorageType.SSD, StorageType.DISK }, { StorageType.SSD, StorageType.DISK }, { StorageType.SSD, StorageType.DISK } }, true, null, null, null, capacities, null, false, false, false, null); cluster.triggerHeartbeats(); client.setStoragePolicy(barDir, "ONE_SSD"); rc = ToolRunner.run(conf, new Mover.Cli(), new String[] { "-p", barDir }); locatedBlocks = client.getBlockLocations(fooFile, 0, fileLen); for (LocatedBlock lb : locatedBlocks.getLocatedBlocks()) { for (StorageType type : lb.getStorageTypes()) { Assert.assertEquals(StorageType.ARCHIVE, type); } } }finally{ cluster.shutdown(); } }
@Test(timeout = 90000) public void testMoverWithPinnedBlocks() throws Exception { final Configuration conf = new HdfsConfiguration(); initConf(conf); conf.setInt(DFSConfigKeys.DFS_MOVER_RETRY_MAX_ATTEMPTS_KEY, 10000); final MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf) .numDataNodes(3) .build(); try { cluster.waitActive(); final DistributedFileSystem dfs = cluster.getFileSystem(); final String file = "/testMoverWithPinnedBlocks/file"; Path dir = new Path("/testMoverWithPinnedBlocks"); dfs.mkdirs(dir); dfs.setStoragePolicy(dir, "HOT"); final FSDataOutputStream out = dfs.create(new Path(file)); byte[] fileData = StripedFileTestUtil .generateBytes(DEFAULT_BLOCK_SIZE * 3); out.write(fileData); out.close(); LocatedBlock lb = dfs.getClient().getLocatedBlocks(file, 0).get(0); StorageType[] storageTypes = lb.getStorageTypes(); for (StorageType storageType : storageTypes) { Assert.assertTrue(StorageType.DISK == storageType); } StorageType[][] newtypes = new StorageType[][] {{StorageType.SSD}}; startAdditionalDNs(conf, 1, newtypes, cluster); for (int i = 0; i < cluster.getDataNodes().size(); i++) { DataNode dn = cluster.getDataNodes().get(i); LOG.info("Simulate block pinning in datanode {}", dn); InternalDataNodeTestUtils.mockDatanodeBlkPinning(dn, true); } dfs.setStoragePolicy(dir, "ONE_SSD"); int rc = ToolRunner.run(conf, new Mover.Cli(), new String[] {"-p", dir.toString()}); int exitcode = ExitStatus.NO_MOVE_BLOCK.getExitCode(); Assert.assertEquals("Movement should fail", exitcode, rc); } finally { cluster.shutdown(); } }
@Test(timeout = 90000) public void testMoverFailedRetryWithPinnedBlocks() throws Exception { final Configuration conf = new HdfsConfiguration(); initConf(conf); conf.set(DFSConfigKeys.DFS_MOVER_RETRY_MAX_ATTEMPTS_KEY, "2"); final MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf) .numDataNodes(2) .storageTypes( new StorageType[][] {{StorageType.DISK, StorageType.ARCHIVE}, {StorageType.DISK, StorageType.ARCHIVE}}).build(); try { cluster.waitActive(); final DistributedFileSystem dfs = cluster.getFileSystem(); final String parenDir = "/parent"; dfs.mkdirs(new Path(parenDir)); final String file1 = "/parent/testMoverFailedRetryWithPinnedBlocks1"; final FSDataOutputStream out = dfs.create(new Path(file1), (short) 2); byte[] fileData = StripedFileTestUtil .generateBytes(DEFAULT_BLOCK_SIZE * 2); out.write(fileData); out.close(); createFileWithFavoredDatanodes(conf, cluster, dfs); LocatedBlocks locatedBlocks = dfs.getClient().getLocatedBlocks(file1, 0); Assert.assertEquals("Wrong block count", 2, locatedBlocks.locatedBlockCount()); LocatedBlock lb = locatedBlocks.get(0); cluster.corruptBlockOnDataNodesByDeletingBlockFile(lb.getBlock()); dfs.setStoragePolicy(new Path(parenDir), "COLD"); int rc = ToolRunner.run(conf, new Mover.Cli(), new String[] {"-p", parenDir.toString()}); Assert.assertEquals("Movement should fail after some retry", ExitStatus.NO_MOVE_PROGRESS.getExitCode(), rc); } finally { cluster.shutdown(); } }
@Test(timeout = 300000) public void testMoverWhenStoragePolicyUnset() throws Exception { final Configuration conf = new HdfsConfiguration(); initConf(conf); final MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf) .numDataNodes(1) .storageTypes( new StorageType[][] {{StorageType.DISK, StorageType.ARCHIVE}}) .build(); try { cluster.waitActive(); final DistributedFileSystem dfs = cluster.getFileSystem(); final String file = "/testMoverWhenStoragePolicyUnset"; DFSTestUtil.createFile(dfs, new Path(file), 1L, (short) 1, 0L); dfs.setStoragePolicy(new Path(file), "COLD"); int rc = ToolRunner.run(conf, new Mover.Cli(), new String[] {"-p", file.toString()}); Assert.assertEquals("Movement to ARCHIVE should be successful", 0, rc); waitForLocatedBlockWithArchiveStorageType(dfs, file, 1); LocatedBlock lb = dfs.getClient().getLocatedBlocks(file, 0).get(0); Assert.assertTrue(StorageType.ARCHIVE == (lb.getStorageTypes())[0]); dfs.unsetStoragePolicy(new Path(file)); rc = ToolRunner.run(conf, new Mover.Cli(), new String[] {"-p", file.toString()}); Assert.assertEquals("Movement to DISK should be successful", 0, rc); lb = dfs.getClient().getLocatedBlocks(file, 0).get(0); Assert.assertTrue(StorageType.DISK == (lb.getStorageTypes())[0]); } finally { cluster.shutdown(); } }