Balancer { static int run(Collection<URI> namenodes, final BalancerParameters 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); LOG.info("included nodes = " + p.getIncludedNodes()); LOG.info("excluded nodes = " + p.getExcludedNodes()); LOG.info("source nodes = " + p.getSourceNodes()); 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.getMaxIdleIteration()); boolean done = false; for(int iteration = 0; !done; iteration++) { done = true; Collections.shuffle(connectors); for(NameNodeConnector nnc : connectors) { if (p.getBlockPools().size() == 0 || p.getBlockPools().contains(nnc.getBlockpoolID())) { 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); } } else { LOG.info("Skipping blockpool " + nnc.getBlockpoolID()); } } } } finally { for(NameNodeConnector nnc : connectors) { IOUtils.cleanup(LOG, nnc); } } return ExitStatus.SUCCESS.getExitCode(); } Balancer(NameNodeConnector theblockpool, BalancerParameters 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[] hosts = {"host0", "host1"}; String[] racks = { RACK0, RACK1 }; int numOfDatanodes = capacities.length; cluster = new MiniDFSCluster.Builder(conf).numDataNodes(capacities.length) .hosts(hosts).racks(racks).simulatedCapacities(capacities).build(); 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++) { int port = cluster.getDataNodes().get(i).getXferAddress().getPort(); favoredNodes[i] = new InetSocketAddress(hosts[i], port); } 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, BalancerParameters.DEFAULT, conf); assertEquals(ExitStatus.NO_MOVE_PROGRESS.getExitCode(), r); }
@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(); 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()); BalancerParameters.Builder pBuilder = new BalancerParameters.Builder(); pBuilder.setExcludedNodes(datanodes); pBuilder.setRunDuringUpgrade(false); final int r = Balancer.run(namenodes, pBuilder.build(), conf); assertEquals(ExitStatus.SUCCESS.getExitCode(), r); }
@Test(timeout=300000) public void testBalancerWithRamDisk() throws Exception { final int SEED = 0xFADED; final short REPL_FACT = 1; Configuration conf = new Configuration(); 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); initConfWithRamDisk(conf, ramDiskStorageLimit); cluster = new MiniDFSCluster .Builder(conf) .numDataNodes(1) .storageCapacities(new long[] { ramDiskStorageLimit, diskStorageLimit }) .storageTypes(new StorageType[] { RAM_DISK, DEFAULT }) .build(); 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); final BalancerParameters p = BalancerParameters.DEFAULT; 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); }
@Test(timeout=300000) public void testBalancerDuringUpgrade() throws Exception { final int SEED = 0xFADED; Configuration conf = new HdfsConfiguration(); conf.setLong(DFS_HEARTBEAT_INTERVAL_KEY, 1); conf.setInt(DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY, 500); conf.setLong(DFSConfigKeys.DFS_NAMENODE_REPLICATION_INTERVAL_KEY, 1); conf.setLong(DFSConfigKeys.DFS_BALANCER_GETBLOCKS_MIN_BLOCK_SIZE_KEY, 1L); final int BLOCK_SIZE = 1024*1024; cluster = new MiniDFSCluster .Builder(conf) .numDataNodes(1) .storageCapacities(new long[] { BLOCK_SIZE * 10 }) .storageTypes(new StorageType[] { DEFAULT }) .storagesPerDatanode(1) .build(); cluster.waitActive(); final String METHOD_NAME = GenericTestUtils.getMethodName(); final Path path1 = new Path("/" + METHOD_NAME + ".01.dat"); DistributedFileSystem fs = cluster.getFileSystem(); DFSTestUtil.createFile(fs, path1, BLOCK_SIZE, BLOCK_SIZE * 2, BLOCK_SIZE, (short) 1, SEED); cluster.startDataNodes(conf, 1, true, null, null); cluster.triggerHeartbeats(); Collection<URI> namenodes = DFSUtil.getNsServiceRpcUris(conf); final BalancerParameters p = BalancerParameters.DEFAULT; fs.setSafeMode(HdfsConstants.SafeModeAction.SAFEMODE_ENTER); fs.rollingUpgrade(HdfsConstants.RollingUpgradeAction.PREPARE); fs.setSafeMode(HdfsConstants.SafeModeAction.SAFEMODE_LEAVE); assertEquals(ExitStatus.UNFINALIZED_UPGRADE.getExitCode(), Balancer.run(namenodes, p, conf)); BalancerParameters.Builder b = new BalancerParameters.Builder(); b.setRunDuringUpgrade(true); final BalancerParameters runDuringUpgrade = b.build(); assertEquals(ExitStatus.SUCCESS.getExitCode(), Balancer.run(namenodes, runDuringUpgrade, conf)); fs.rollingUpgrade(HdfsConstants.RollingUpgradeAction.FINALIZE); assertEquals(ExitStatus.SUCCESS.getExitCode(), Balancer.run(namenodes, p, conf)); }
@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); conf.setLong(DFSConfigKeys.DFS_BALANCER_GETBLOCKS_MIN_BLOCK_SIZE_KEY, 1L); int numOfDatanodes =2; 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(); 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(); BalancerParameters p = BalancerParameters.DEFAULT; Collection<URI> namenodes = DFSUtil.getNsServiceRpcUris(conf); final int r = Balancer.run(namenodes, p, conf); assertEquals(ExitStatus.NO_MOVE_PROGRESS.getExitCode(), r); }
@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(); 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); }
@Test(timeout=60000) public void testMinBlockSizeAndSourceNodes() throws Exception { final Configuration conf = new HdfsConfiguration(); initConf(conf); final short replication = 3; final long[] lengths = {10, 10, 10, 10}; final long[] capacities = new long[replication]; final long totalUsed = capacities.length * sum(lengths); Arrays.fill(capacities, 1000); cluster = new MiniDFSCluster.Builder(conf) .numDataNodes(capacities.length) .simulatedCapacities(capacities) .build(); final DistributedFileSystem dfs = cluster.getFileSystem(); cluster.waitActive(); client = NameNodeProxies.createProxy(conf, dfs.getUri(), ClientProtocol.class).getProxy(); for(int i = 0; i < lengths.length; i++) { final long size = lengths[i]; final Path p = new Path("/file" + i + "_size" + size); try(final OutputStream out = dfs.create(p)) { for(int j = 0; j < size; j++) { out.write(j); } } } cluster.startDataNodes(conf, capacities.length, true, null, null, capacities); LOG.info("capacities = " + Arrays.toString(capacities)); LOG.info("totalUsedSpace= " + totalUsed); LOG.info("lengths = " + Arrays.toString(lengths) + ", #=" + lengths.length); waitForHeartBeat(totalUsed, 2*capacities[0]*capacities.length, client, cluster); final Collection<URI> namenodes = DFSUtil.getNsServiceRpcUris(conf); { BalancerParameters.Builder b = new BalancerParameters.Builder(); b.setBalancingPolicy(BalancingPolicy.Node.INSTANCE); b.setThreshold(1); final BalancerParameters p = b.build(); conf.setLong(DFSConfigKeys.DFS_BALANCER_GETBLOCKS_MIN_BLOCK_SIZE_KEY, 50); final int r = Balancer.run(namenodes, p, conf); assertEquals(ExitStatus.NO_MOVE_PROGRESS.getExitCode(), r); } conf.setLong(DFSConfigKeys.DFS_BALANCER_GETBLOCKS_MIN_BLOCK_SIZE_KEY, 1); { final Set<String> sourceNodes = new HashSet<>(); final List<DataNode> datanodes = cluster.getDataNodes(); for(int i = capacities.length; i < datanodes.size(); i++) { sourceNodes.add(datanodes.get(i).getDisplayName()); } BalancerParameters.Builder b = new BalancerParameters.Builder(); b.setBalancingPolicy(BalancingPolicy.Node.INSTANCE); b.setThreshold(1); b.setSourceNodes(sourceNodes); final BalancerParameters p = b.build(); conf.setLong(DFSConfigKeys.DFS_BALANCER_GETBLOCKS_MIN_BLOCK_SIZE_KEY, 50); final int r = Balancer.run(namenodes, p, conf); assertEquals(ExitStatus.NO_MOVE_BLOCK.getExitCode(), r); } { final Set<String> sourceNodes = new HashSet<>(); final List<DataNode> datanodes = cluster.getDataNodes(); sourceNodes.add(datanodes.get(0).getDisplayName()); BalancerParameters.Builder b = new BalancerParameters.Builder(); b.setBalancingPolicy(BalancingPolicy.Node.INSTANCE); b.setThreshold(1); b.setSourceNodes(sourceNodes); final BalancerParameters p = b.build(); conf.setLong(DFSConfigKeys.DFS_BALANCER_GETBLOCKS_MIN_BLOCK_SIZE_KEY, 1); final int r = Balancer.run(namenodes, p, conf); assertEquals(ExitStatus.NO_MOVE_BLOCK.getExitCode(), r); } { final Set<String> sourceNodes = new HashSet<>(); final List<DataNode> datanodes = cluster.getDataNodes(); for(int i = 0; i < capacities.length; i++) { sourceNodes.add(datanodes.get(i).getDisplayName()); } BalancerParameters.Builder b = new BalancerParameters.Builder(); b.setBalancingPolicy(BalancingPolicy.Node.INSTANCE); b.setThreshold(1); b.setSourceNodes(sourceNodes); final BalancerParameters p = b.build(); conf.setLong(DFSConfigKeys.DFS_BALANCER_GETBLOCKS_MIN_BLOCK_SIZE_KEY, 1); final int r = Balancer.run(namenodes, p, conf); assertEquals(ExitStatus.SUCCESS.getExitCode(), r); } }