Balancer { static int run(Collection<URI> namenodes, final Parameters p, Configuration conf) throws IOException, InterruptedException { final long sleeptime = conf.getLong(DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_KEY, DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_DEFAULT) * 2000 + conf.getLong(DFSConfigKeys.DFS_NAMENODE_REPLICATION_INTERVAL_KEY, DFSConfigKeys.DFS_NAMENODE_REPLICATION_INTERVAL_DEFAULT) * 1000; LOG.info("namenodes = " + namenodes); LOG.info("parameters = " + p); System.out.println("Time Stamp Iteration# Bytes Already Moved Bytes Left To Move Bytes Being Moved"); List<NameNodeConnector> connectors = Collections.emptyList(); try { connectors = NameNodeConnector.newNameNodeConnectors(namenodes, Balancer.class.getSimpleName(), BALANCER_ID_PATH, conf, p.maxIdleIteration); boolean done = false; for(int iteration = 0; !done; iteration++) { done = true; Collections.shuffle(connectors); for(NameNodeConnector nnc : connectors) { final Balancer b = new Balancer(nnc, p, conf); final Result r = b.runOneIteration(); r.print(iteration, System.out); b.resetData(conf); if (r.exitStatus == ExitStatus.IN_PROGRESS) { done = false; } else if (r.exitStatus != ExitStatus.SUCCESS) { return r.exitStatus.getExitCode(); } } if (!done) { Thread.sleep(sleeptime); } } } finally { for(NameNodeConnector nnc : connectors) { IOUtils.cleanup(LOG, nnc); } } return ExitStatus.SUCCESS.getExitCode(); } Balancer(NameNodeConnector theblockpool, Parameters p, Configuration conf); static void main(String[] args);  }
@Test(timeout=100000) public void testBalancerWithPinnedBlocks() throws Exception { assumeTrue(!Path.WINDOWS); final Configuration conf = new HdfsConfiguration(); initConf(conf); conf.setBoolean(DFS_DATANODE_BLOCK_PINNING_ENABLED, true); long[] capacities = new long[] { CAPACITY, CAPACITY }; String[] racks = { RACK0, RACK1 }; int numOfDatanodes = capacities.length; cluster = new MiniDFSCluster.Builder(conf).numDataNodes(capacities.length) .hosts(new String[]{"localhost", "localhost"}) .racks(racks).simulatedCapacities(capacities).build(); try { cluster.waitActive(); client = NameNodeProxies.createProxy(conf, cluster.getFileSystem(0).getUri(), ClientProtocol.class).getProxy(); long totalCapacity = sum(capacities); long totalUsedSpace = totalCapacity * 8 / 10; InetSocketAddress[] favoredNodes = new InetSocketAddress[numOfDatanodes]; for (int i = 0; i < favoredNodes.length; i++) { favoredNodes[i] = cluster.getDataNodes().get(i).getXferAddress(); } DFSTestUtil.createFile(cluster.getFileSystem(0), filePath, 1024, totalUsedSpace / numOfDatanodes, DEFAULT_BLOCK_SIZE, (short) numOfDatanodes, 0, false, favoredNodes); cluster.startDataNodes(conf, 1, true, null, new String[] { RACK2 }, new long[] { CAPACITY }); totalCapacity += CAPACITY; waitForHeartBeat(totalUsedSpace, totalCapacity, client, cluster); Collection<URI> namenodes = DFSUtil.getNsServiceRpcUris(conf); int r = Balancer.run(namenodes, Balancer.Parameters.DEFAULT, conf); assertEquals(ExitStatus.NO_MOVE_PROGRESS.getExitCode(), r); } finally { cluster.shutdown(); } }
@Test(timeout=300000) public void testUnknownDatanode() throws IOException, InterruptedException, TimeoutException, URISyntaxException { Configuration conf = new HdfsConfiguration(); initConf(conf); long distribution[] = new long[] {50*CAPACITY/100, 70*CAPACITY/100, 0*CAPACITY/100}; long capacities[] = new long[]{CAPACITY, CAPACITY, CAPACITY}; String racks[] = new String[] {RACK0, RACK1, RACK1}; int numDatanodes = distribution.length; if (capacities.length != numDatanodes || racks.length != numDatanodes) { throw new IllegalArgumentException("Array length is not the same"); } final long totalUsedSpace = sum(distribution); ExtendedBlock[] blocks = generateBlocks(conf, totalUsedSpace, (short) numDatanodes); Block[][] blocksDN = distributeBlocks( blocks, (short)(numDatanodes-1), distribution); conf.set(DFSConfigKeys.DFS_NAMENODE_SAFEMODE_THRESHOLD_PCT_KEY, "0.0f"); cluster = new MiniDFSCluster.Builder(conf).numDataNodes(numDatanodes) .format(false) .racks(racks) .simulatedCapacities(capacities) .build(); cluster.waitActive(); client = NameNodeProxies.createProxy(conf, cluster.getFileSystem(0).getUri(), ClientProtocol.class).getProxy(); Thread.sleep(2000); for(int i = 0; i < 3; i++) { cluster.injectBlocks(i, Arrays.asList(blocksDN[i]), null); } cluster.startDataNodes(conf, 1, true, null, new String[]{RACK0}, null,new long[]{CAPACITY}); cluster.triggerHeartbeats(); Collection<URI> namenodes = DFSUtil.getNsServiceRpcUris(conf); Set<String> datanodes = new HashSet<String>(); datanodes.add(cluster.getDataNodes().get(0).getDatanodeId().getHostName()); Balancer.Parameters p = new Balancer.Parameters( Balancer.Parameters.DEFAULT.policy, Balancer.Parameters.DEFAULT.threshold, Balancer.Parameters.DEFAULT.maxIdleIteration, datanodes, Balancer.Parameters.DEFAULT.nodesToBeIncluded); final int r = Balancer.run(namenodes, p, conf); assertEquals(ExitStatus.SUCCESS.getExitCode(), r); }
@Test(timeout=100000) public void testTwoReplicaShouldNotInSameDN() throws Exception { final Configuration conf = new HdfsConfiguration(); int blockSize = 5 * 1024 * 1024 ; conf.setLong(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, blockSize); conf.setLong(DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_KEY, 1L); conf.setLong(DFSConfigKeys.DFS_NAMENODE_REPLICATION_INTERVAL_KEY, 1L); int numOfDatanodes =2; final MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf) .numDataNodes(2) .racks(new String[]{"/default/rack0", "/default/rack0"}) .storagesPerDatanode(2) .storageTypes(new StorageType[][]{ {StorageType.SSD, StorageType.DISK}, {StorageType.SSD, StorageType.DISK}}) .storageCapacities(new long[][]{ {100 * blockSize, 20 * blockSize}, {20 * blockSize, 100 * blockSize}}) .build(); try { cluster.waitActive(); DistributedFileSystem fs = cluster.getFileSystem(); Path barDir = new Path("/bar"); fs.mkdir(barDir,new FsPermission((short)777)); fs.setStoragePolicy(barDir, HdfsConstants.ONESSD_STORAGE_POLICY_NAME); long fileLen = 30 * blockSize; Path fooFile = new Path(barDir, "foo"); createFile(cluster, fooFile, fileLen, (short) numOfDatanodes, 0); cluster.triggerHeartbeats(); Balancer.Parameters p = Balancer.Parameters.DEFAULT; Collection<URI> namenodes = DFSUtil.getNsServiceRpcUris(conf); final int r = Balancer.run(namenodes, p, conf); assertEquals(ExitStatus.NO_MOVE_PROGRESS.getExitCode(), r); } finally { cluster.shutdown(); } }
@Test(timeout=100000) public void testBalancerWithPinnedBlocks() throws Exception { assumeTrue(!Path.WINDOWS); final Configuration conf = new HdfsConfiguration(); initConf(conf); conf.setBoolean(DFS_DATANODE_BLOCK_PINNING_ENABLED, true); long[] capacities = new long[] { CAPACITY, CAPACITY }; String[] racks = { RACK0, RACK1 }; int numOfDatanodes = capacities.length; cluster = new MiniDFSCluster.Builder(conf).numDataNodes(capacities.length) .hosts(new String[]{"localhost", "localhost"}) .racks(racks).simulatedCapacities(capacities).build(); try { cluster.waitActive(); client = NameNodeProxies.createProxy(conf, cluster.getFileSystem(0).getUri(), ClientProtocol.class).getProxy(); long totalCapacity = sum(capacities); long totalUsedSpace = totalCapacity * 8 / 10; InetSocketAddress[] favoredNodes = new InetSocketAddress[numOfDatanodes]; for (int i = 0; i < favoredNodes.length; i++) { favoredNodes[i] = cluster.getDataNodes().get(i).getXferAddress(); } DFSTestUtil.createFile(cluster.getFileSystem(0), filePath, false, 1024, totalUsedSpace / numOfDatanodes, DEFAULT_BLOCK_SIZE, (short) numOfDatanodes, 0, false, favoredNodes); cluster.startDataNodes(conf, 1, true, null, new String[] { RACK2 }, new long[] { CAPACITY }); totalCapacity += CAPACITY; waitForHeartBeat(totalUsedSpace, totalCapacity, client, cluster); Collection<URI> namenodes = DFSUtil.getNsServiceRpcUris(conf); int r = Balancer.run(namenodes, Balancer.Parameters.DEFAULT, conf); assertEquals(ExitStatus.NO_MOVE_PROGRESS.getExitCode(), r); } finally { cluster.shutdown(); } }
@Test(timeout=100000) public void testUnknownDatanode() throws Exception { Configuration conf = new HdfsConfiguration(); initConf(conf); long distribution[] = new long[] {50*CAPACITY/100, 70*CAPACITY/100, 0*CAPACITY/100}; long capacities[] = new long[]{CAPACITY, CAPACITY, CAPACITY}; String racks[] = new String[] {RACK0, RACK1, RACK1}; int numDatanodes = distribution.length; if (capacities.length != numDatanodes || racks.length != numDatanodes) { throw new IllegalArgumentException("Array length is not the same"); } final long totalUsedSpace = sum(distribution); ExtendedBlock[] blocks = generateBlocks(conf, totalUsedSpace, (short) numDatanodes); Block[][] blocksDN = distributeBlocks( blocks, (short)(numDatanodes-1), distribution); conf.set(DFSConfigKeys.DFS_NAMENODE_SAFEMODE_THRESHOLD_PCT_KEY, "0.0f"); cluster = new MiniDFSCluster.Builder(conf).numDataNodes(numDatanodes) .format(false) .racks(racks) .simulatedCapacities(capacities) .build(); try { cluster.waitActive(); client = NameNodeProxies.createProxy(conf, cluster.getFileSystem(0).getUri(), ClientProtocol.class).getProxy(); for(int i = 0; i < 3; i++) { cluster.injectBlocks(i, Arrays.asList(blocksDN[i]), null); } cluster.startDataNodes(conf, 1, true, null, new String[]{RACK0}, null,new long[]{CAPACITY}); cluster.triggerHeartbeats(); Collection<URI> namenodes = DFSUtil.getNsServiceRpcUris(conf); Set<String> datanodes = new HashSet<String>(); datanodes.add(cluster.getDataNodes().get(0).getDatanodeId().getHostName()); Balancer.Parameters p = new Balancer.Parameters( Balancer.Parameters.DEFAULT.policy, Balancer.Parameters.DEFAULT.threshold, Balancer.Parameters.DEFAULT.maxIdleIteration, datanodes, Balancer.Parameters.DEFAULT.nodesToBeIncluded); final int r = Balancer.run(namenodes, p, conf); assertEquals(ExitStatus.SUCCESS.getExitCode(), r); } finally { cluster.shutdown(); } }
@Test(timeout=300000) public void testBalancerWithRamDisk() throws Exception { final int SEED = 0xFADED; final short REPL_FACT = 1; Configuration conf = new Configuration(); initConfWithRamDisk(conf); final int defaultRamDiskCapacity = 10; final long ramDiskStorageLimit = ((long) defaultRamDiskCapacity * DEFAULT_RAM_DISK_BLOCK_SIZE) + (DEFAULT_RAM_DISK_BLOCK_SIZE - 1); final long diskStorageLimit = ((long) defaultRamDiskCapacity * DEFAULT_RAM_DISK_BLOCK_SIZE) + (DEFAULT_RAM_DISK_BLOCK_SIZE - 1); cluster = new MiniDFSCluster .Builder(conf) .numDataNodes(1) .storageCapacities(new long[] { ramDiskStorageLimit, diskStorageLimit }) .storageTypes(new StorageType[] { RAM_DISK, DEFAULT }) .build(); try { cluster.waitActive(); final String METHOD_NAME = GenericTestUtils.getMethodName(); final Path path1 = new Path("/" + METHOD_NAME + ".01.dat"); final Path path2 = new Path("/" + METHOD_NAME + ".02.dat"); DistributedFileSystem fs = cluster.getFileSystem(); DFSClient client = fs.getClient(); DFSTestUtil.createFile(fs, path1, true, DEFAULT_RAM_DISK_BLOCK_SIZE, 4 * DEFAULT_RAM_DISK_BLOCK_SIZE, DEFAULT_RAM_DISK_BLOCK_SIZE, REPL_FACT, SEED, true); DFSTestUtil.createFile(fs, path2, true, DEFAULT_RAM_DISK_BLOCK_SIZE, 1 * DEFAULT_RAM_DISK_BLOCK_SIZE, DEFAULT_RAM_DISK_BLOCK_SIZE, REPL_FACT, SEED, true); Thread.sleep(6 * 1000); StorageType[][] storageTypes = new StorageType[][] {{RAM_DISK, DEFAULT}}; long[][] storageCapacities = new long[][]{{ramDiskStorageLimit, diskStorageLimit}}; cluster.startDataNodes(conf, REPL_FACT, storageTypes, true, null, null, null, storageCapacities, null, false, false, false, null); cluster.triggerHeartbeats(); Collection<URI> namenodes = DFSUtil.getNsServiceRpcUris(conf); Balancer.Parameters p = new Balancer.Parameters( Parameters.DEFAULT.policy, Parameters.DEFAULT.threshold, Balancer.Parameters.DEFAULT.maxIdleIteration, Parameters.DEFAULT.nodesToBeExcluded, Parameters.DEFAULT.nodesToBeIncluded); final int r = Balancer.run(namenodes, p, conf); assertEquals(ExitStatus.NO_MOVE_PROGRESS.getExitCode(), r); DFSTestUtil.verifyFileReplicasOnStorageType(fs, client, path1, RAM_DISK); DFSTestUtil.verifyFileReplicasOnStorageType(fs, client, path2, RAM_DISK); } finally { cluster.shutdown(); } }
@Test(timeout = 100000) public void testManyBalancerSimultaneously() throws Exception { final Configuration conf = new HdfsConfiguration(); initConf(conf); long[] capacities = new long[] { 4 * CAPACITY }; String[] racks = new String[] { RACK0 }; long newCapacity = 2 * CAPACITY; String newRack = RACK0; LOG.info("capacities = " + long2String(capacities)); LOG.info("racks = " + Arrays.asList(racks)); LOG.info("newCapacity= " + newCapacity); LOG.info("newRack = " + newRack); LOG.info("useTool = " + false); assertEquals(capacities.length, racks.length); int numOfDatanodes = capacities.length; cluster = new MiniDFSCluster.Builder(conf).numDataNodes(capacities.length) .racks(racks).simulatedCapacities(capacities).build(); try { cluster.waitActive(); client = NameNodeProxies.createProxy(conf, cluster.getFileSystem(0).getUri(), ClientProtocol.class).getProxy(); long totalCapacity = sum(capacities); final long totalUsedSpace = totalCapacity * 3 / 10; createFile(cluster, filePath, totalUsedSpace / numOfDatanodes, (short) numOfDatanodes, 0); cluster.startDataNodes(conf, 1, true, null, new String[] { newRack }, new long[] { newCapacity }); FileSystem fs = cluster.getFileSystem(0); final FSDataOutputStream out = fs .create(Balancer.BALANCER_ID_PATH, false); out.writeBytes(InetAddress.getLocalHost().getHostName()); out.hflush(); assertTrue("'balancer.id' file doesn't exist!", fs.exists(Balancer.BALANCER_ID_PATH)); final String[] args = { "-policy", "datanode" }; final Tool tool = new Cli(); tool.setConf(conf); int exitCode = tool.run(args); assertEquals("Exit status code mismatches", ExitStatus.IO_EXCEPTION.getExitCode(), exitCode); out.close(); assertTrue("'balancer.id' file doesn't exist!", fs.exists(Balancer.BALANCER_ID_PATH)); exitCode = tool.run(args); assertEquals("Exit status code mismatches", ExitStatus.SUCCESS.getExitCode(), exitCode); } finally { cluster.shutdown(); } }