Balancer { static int run(Collection<URI> namenodes, final BalancerParameters p, Configuration conf) throws IOException, InterruptedException { return run(namenodes, null, p, conf); } Balancer(NameNodeConnector theblockpool, BalancerParameters p,
      Configuration conf); static void main(String[] args);  }
@Test(timeout=100000) public void testBalancerWithPinnedBlocks() throws Exception { assumeNotWindows(); 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.getInternalNsRpcUris(conf); int r = Balancer.run(namenodes, BalancerParameters.DEFAULT, conf); assertEquals(ExitStatus.NO_MOVE_PROGRESS.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.getInternalNsRpcUris(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_REDUNDANCY_INTERVAL_SECONDS_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.getInternalNsRpcUris(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_REDUNDANCY_INTERVAL_SECONDS_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.getInternalNsRpcUris(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 }); cluster.triggerHeartbeats(); 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(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.getInternalNsRpcUris(conf); { final BalancerParameters p = Balancer.Cli.parse(new String[] { "-policy", BalancingPolicy.Node.INSTANCE.getName(), "-threshold", "1" }); assertEquals(p.getBalancingPolicy(), BalancingPolicy.Node.INSTANCE); assertEquals(p.getThreshold(), 1.0, 0.001); 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()); } final BalancerParameters p = Balancer.Cli.parse(new String[] { "-policy", BalancingPolicy.Node.INSTANCE.getName(), "-threshold", "1", "-source", StringUtils.join(sourceNodes, ',') }); assertEquals(p.getBalancingPolicy(), BalancingPolicy.Node.INSTANCE); assertEquals(p.getThreshold(), 1.0, 0.001); assertEquals(p.getSourceNodes(), sourceNodes); 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()); final BalancerParameters p = Balancer.Cli.parse(new String[] { "-policy", BalancingPolicy.Node.INSTANCE.getName(), "-threshold", "1", "-source", StringUtils.join(sourceNodes, ',') }); assertEquals(p.getBalancingPolicy(), BalancingPolicy.Node.INSTANCE); assertEquals(p.getThreshold(), 1.0, 0.001); assertEquals(p.getSourceNodes(), sourceNodes); 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()); } final BalancerParameters p = Balancer.Cli.parse(new String[] { "-policy", BalancingPolicy.Node.INSTANCE.getName(), "-threshold", "1", "-source", StringUtils.join(sourceNodes, ',') }); assertEquals(p.getBalancingPolicy(), BalancingPolicy.Node.INSTANCE); assertEquals(p.getThreshold(), 1.0, 0.001); assertEquals(p.getSourceNodes(), sourceNodes); conf.setLong(DFSConfigKeys.DFS_BALANCER_GETBLOCKS_MIN_BLOCK_SIZE_KEY, 1); final int r = Balancer.run(namenodes, p, conf); assertEquals(ExitStatus.SUCCESS.getExitCode(), r); } }