http://git-wip-us.apache.org/repos/asf/hadoop/blob/e5bd5714/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/sps/TestExternalStoragePolicySatisfier.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/sps/TestExternalStoragePolicySatisfier.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/sps/TestExternalStoragePolicySatisfier.java index 18acb50..d9a93fd 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/sps/TestExternalStoragePolicySatisfier.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/sps/TestExternalStoragePolicySatisfier.java @@ -32,34 +32,57 @@ import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_SPS_ADDRESS_KEY; import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_SPS_KERBEROS_PRINCIPAL_KEY; import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_SPS_KEYTAB_FILE_KEY; import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_SPS_MAX_OUTSTANDING_PATHS_KEY; +import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_STORAGE_POLICY_ENABLED_KEY; import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_WEB_AUTHENTICATION_KERBEROS_PRINCIPAL_KEY; import static org.apache.hadoop.hdfs.client.HdfsClientConfigKeys.DFS_DATA_TRANSFER_PROTECTION_KEY; +import static org.apache.hadoop.hdfs.server.common.HdfsServerConstants.XATTR_SATISFY_STORAGE_POLICY; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.fail; import java.io.File; +import java.io.FileNotFoundException; import java.io.IOException; -import java.net.URI; +import java.net.InetSocketAddress; import java.security.PrivilegedExceptionAction; import java.util.ArrayList; -import java.util.Collection; import java.util.List; import java.util.Properties; import java.util.concurrent.TimeoutException; +import org.apache.commons.logging.LogFactory; import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FSDataOutputStream; +import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.FileUtil; import org.apache.hadoop.fs.Path; import org.apache.hadoop.fs.StorageType; +import org.apache.hadoop.fs.permission.FsPermission; import org.apache.hadoop.hdfs.DFSConfigKeys; import org.apache.hadoop.hdfs.DFSTestUtil; -import org.apache.hadoop.hdfs.DFSUtil; import org.apache.hadoop.hdfs.DistributedFileSystem; +import org.apache.hadoop.hdfs.HdfsConfiguration; import org.apache.hadoop.hdfs.MiniDFSCluster; +import org.apache.hadoop.hdfs.NameNodeProxies; +import org.apache.hadoop.hdfs.StripedFileTestUtil; +import org.apache.hadoop.hdfs.MiniDFSCluster.DataNodeProperties; +import org.apache.hadoop.hdfs.client.HdfsAdmin; +import org.apache.hadoop.hdfs.protocol.Block; +import org.apache.hadoop.hdfs.protocol.ClientProtocol; +import org.apache.hadoop.hdfs.protocol.DatanodeInfo; +import org.apache.hadoop.hdfs.protocol.HdfsConstants; +import org.apache.hadoop.hdfs.protocol.LocatedBlock; +import org.apache.hadoop.hdfs.protocol.LocatedBlocks; import org.apache.hadoop.hdfs.protocol.HdfsConstants.StoragePolicySatisfierMode; import org.apache.hadoop.hdfs.server.balancer.NameNodeConnector; import org.apache.hadoop.hdfs.server.common.HdfsServerConstants; +import org.apache.hadoop.hdfs.server.datanode.DataNode; +import org.apache.hadoop.hdfs.server.datanode.DataNodeTestUtils; +import org.apache.hadoop.hdfs.server.datanode.InternalDataNodeTestUtils; +import org.apache.hadoop.hdfs.server.namenode.FSEditLog; +import org.apache.hadoop.hdfs.server.namenode.INode; +import org.apache.hadoop.hdfs.server.namenode.sps.BlockMovementListener; import org.apache.hadoop.hdfs.server.namenode.sps.BlockStorageMovementAttemptedItems; import org.apache.hadoop.hdfs.server.namenode.sps.StoragePolicySatisfier; -import org.apache.hadoop.hdfs.server.namenode.sps.TestStoragePolicySatisfier; import org.apache.hadoop.http.HttpConfig; import org.apache.hadoop.minikdc.MiniKdc; import org.apache.hadoop.security.SecurityUtil; @@ -67,29 +90,57 @@ import org.apache.hadoop.security.UserGroupInformation; import org.apache.hadoop.security.authentication.util.KerberosName; import org.apache.hadoop.security.ssl.KeyStoreTestUtil; import org.apache.hadoop.test.GenericTestUtils; +import org.apache.hadoop.test.GenericTestUtils.LogCapturer; import org.junit.After; import org.junit.Assert; +import org.junit.Before; import org.junit.Ignore; import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import com.google.common.base.Supplier; /** * Tests the external sps service plugins. */ -public class TestExternalStoragePolicySatisfier - extends TestStoragePolicySatisfier { +public class TestExternalStoragePolicySatisfier { + private static final String ONE_SSD = "ONE_SSD"; + private static final String COLD = "COLD"; private StorageType[][] allDiskTypes = new StorageType[][]{{StorageType.DISK, StorageType.DISK}, {StorageType.DISK, StorageType.DISK}, {StorageType.DISK, StorageType.DISK}}; - private NameNodeConnector nnc; private File keytabFile; private String principal; private MiniKdc kdc; private File baseDir; + private NameNodeConnector nnc; private StoragePolicySatisfier externalSps; private ExternalSPSContext externalCtxt; + private DistributedFileSystem dfs = null; + private MiniDFSCluster hdfsCluster = null; + private Configuration config = null; + private static final int NUM_OF_DATANODES = 3; + private static final int STORAGES_PER_DATANODE = 2; + private static final long CAPACITY = 2 * 256 * 1024 * 1024; + private static final String FILE = "/testMoveToSatisfyStoragePolicy"; + private static final int DEFAULT_BLOCK_SIZE = 1024; + private static final Logger LOG = + LoggerFactory.getLogger(TestExternalStoragePolicySatisfier.class); + + @Before + public void setUp() { + config = new HdfsConfiguration(); + config.set(DFSConfigKeys.DFS_STORAGE_POLICY_SATISFIER_MODE_KEY, + StoragePolicySatisfierMode.EXTERNAL.toString()); + // Most of the tests are restarting DNs and NN. So, reduced refresh cycle to + // update latest datanodes. + config.setLong(DFSConfigKeys.DFS_SPS_DATANODE_CACHE_REFRESH_INTERVAL_MS, + 3000); + config.set(DFSConfigKeys.DFS_STORAGE_POLICY_SATISFIER_MODE_KEY, + StoragePolicySatisfierMode.EXTERNAL.toString()); + } @After public void destroy() throws Exception { @@ -97,26 +148,49 @@ public class TestExternalStoragePolicySatisfier kdc.stop(); FileUtil.fullyDelete(baseDir); } + if (hdfsCluster != null) { + hdfsCluster.shutdown(); + } } - @Override - public void shutdownCluster() { - if (externalSps != null) { - externalSps.stopGracefully(); - } - super.shutdownCluster(); + /** + * Sets hdfs cluster. + */ + private void setCluster(MiniDFSCluster cluster) { + this.hdfsCluster = cluster; } - @Override - public void setUp() { - super.setUp(); + /** + * @return conf. + */ + private Configuration getConf() { + return this.config; + } - getConf().set(DFSConfigKeys.DFS_STORAGE_POLICY_SATISFIER_MODE_KEY, - StoragePolicySatisfierMode.EXTERNAL.toString()); + /** + * @return hdfs cluster. + */ + private MiniDFSCluster getCluster() { + return hdfsCluster; + } + + /** + * Gets distributed file system. + * + * @throws IOException + */ + private DistributedFileSystem getFS() throws IOException { + this.dfs = hdfsCluster.getFileSystem(); + return this.dfs; + } + + private void shutdownCluster() { + if (externalSps != null) { + externalSps.stopGracefully(); + } } - @Override - public void createCluster() throws IOException { + private void createCluster() throws IOException { getConf().setLong("dfs.block.size", DEFAULT_BLOCK_SIZE); setCluster(startCluster(getConf(), allDiskTypes, NUM_OF_DATANODES, STORAGES_PER_DATANODE, CAPACITY)); @@ -124,8 +198,7 @@ public class TestExternalStoragePolicySatisfier writeContent(FILE); } - @Override - public MiniDFSCluster startCluster(final Configuration conf, + private MiniDFSCluster startCluster(final Configuration conf, StorageType[][] storageTypes, int numberOfDatanodes, int storagesPerDn, long nodeCapacity) throws IOException { long[][] capacities = new long[numberOfDatanodes][storagesPerDn]; @@ -139,7 +212,8 @@ public class TestExternalStoragePolicySatisfier .storageTypes(storageTypes).storageCapacities(capacities).build(); cluster.waitActive(); - nnc = getNameNodeConnector(getConf()); + nnc = DFSTestUtil.getNameNodeConnector(getConf(), + HdfsServerConstants.MOVER_ID_PATH, 1, false); externalSps = new StoragePolicySatisfier(getConf()); externalCtxt = new ExternalSPSContext(externalSps, nnc); @@ -149,7 +223,7 @@ public class TestExternalStoragePolicySatisfier return cluster; } - public void restartNamenode() throws IOException{ + private void restartNamenode() throws IOException{ if (externalSps != null) { externalSps.stopGracefully(); } @@ -163,60 +237,6 @@ public class TestExternalStoragePolicySatisfier externalSps.start(true, StoragePolicySatisfierMode.EXTERNAL); } - private NameNodeConnector getNameNodeConnector(Configuration conf) - throws IOException { - final Collection<URI> namenodes = DFSUtil.getInternalNsRpcUris(conf); - Assert.assertEquals(1, namenodes.size()); - final Path externalSPSPathId = HdfsServerConstants.MOVER_ID_PATH; - NameNodeConnector.checkOtherInstanceRunning(false); - while (true) { - try { - final List<NameNodeConnector> nncs = NameNodeConnector - .newNameNodeConnectors(namenodes, - StoragePolicySatisfier.class.getSimpleName(), - externalSPSPathId, conf, - NameNodeConnector.DEFAULT_MAX_IDLE_ITERATIONS); - return nncs.get(0); - } catch (IOException e) { - LOG.warn("Failed to connect with namenode", e); - // Ignore - } - - } - } - - public void waitForAttemptedItems(long expectedBlkMovAttemptedCount, - int timeout) throws TimeoutException, InterruptedException { - GenericTestUtils.waitFor(new Supplier<Boolean>() { - @Override - public Boolean get() { - LOG.info("expectedAttemptedItemsCount={} actualAttemptedItemsCount={}", - expectedBlkMovAttemptedCount, - ((BlockStorageMovementAttemptedItems) (externalSps - .getAttemptedItemsMonitor())).getAttemptedItemsCount()); - return ((BlockStorageMovementAttemptedItems) (externalSps - .getAttemptedItemsMonitor())) - .getAttemptedItemsCount() == expectedBlkMovAttemptedCount; - } - }, 100, timeout); - } - - public void waitForBlocksMovementAttemptReport( - long expectedMovementFinishedBlocksCount, int timeout) - throws TimeoutException, InterruptedException { - GenericTestUtils.waitFor(new Supplier<Boolean>() { - @Override - public Boolean get() { - int actualCount = externalSps.getAttemptedItemsMonitor() - .getAttemptedItemsCount(); - LOG.info("MovementFinishedBlocks: expectedCount={} actualCount={}", - expectedMovementFinishedBlocksCount, actualCount); - return actualCount - >= expectedMovementFinishedBlocksCount; - } - }, 100, timeout); - } - private void initSecureConf(Configuration conf) throws Exception { String username = "externalSPS"; baseDir = GenericTestUtils @@ -344,22 +364,6 @@ public class TestExternalStoragePolicySatisfier } /** - * Test verifies status check when Satisfier is not running inside namenode. - */ - @Test(timeout = 90000) - public void testStoragePolicySatisfyPathStatus() throws Exception { - createCluster(); - DistributedFileSystem fs = getFS(); - try { - fs.getClient().checkStoragePolicySatisfyPathStatus(FILE); - Assert.fail("Should throw exception as SPS is not running inside NN!"); - } catch (IOException e) { - GenericTestUtils.assertExceptionContains("Satisfier is not running" - + " inside namenode, so status can't be returned.", e); - } - } - - /** * Tests to verify that SPS should be able to start when the Mover ID file * is not being hold by a Mover. This can be the case when Mover exits * ungracefully without deleting the ID file from HDFS. @@ -399,17 +403,9 @@ public class TestExternalStoragePolicySatisfier } /** - * Status won't be supported for external SPS, now. So, ignoring it. - */ - @Ignore("Status is not supported for external SPS. So, ignoring it.") - public void testMaxRetryForFailedBlock() throws Exception { - } - - /** * This test is specific to internal SPS. So, ignoring it. */ @Ignore("This test is specific to internal SPS. So, ignoring it.") - @Override public void testTraverseWhenParentDeleted() throws Exception { } @@ -417,7 +413,1238 @@ public class TestExternalStoragePolicySatisfier * This test is specific to internal SPS. So, ignoring it. */ @Ignore("This test is specific to internal SPS. So, ignoring it.") - @Override public void testTraverseWhenRootParentDeleted() throws Exception { } + + + @Test(timeout = 300000) + public void testWhenStoragePolicySetToCOLD() + throws Exception { + + try { + createCluster(); + doTestWhenStoragePolicySetToCOLD(); + } finally { + shutdownCluster(); + } + } + + private void doTestWhenStoragePolicySetToCOLD() throws Exception { + // Change policy to COLD + dfs.setStoragePolicy(new Path(FILE), COLD); + + StorageType[][] newtypes = + new StorageType[][]{{StorageType.ARCHIVE, StorageType.ARCHIVE}, + {StorageType.ARCHIVE, StorageType.ARCHIVE}, + {StorageType.ARCHIVE, StorageType.ARCHIVE}}; + startAdditionalDNs(config, 3, NUM_OF_DATANODES, newtypes, + STORAGES_PER_DATANODE, CAPACITY, hdfsCluster); + + hdfsCluster.triggerHeartbeats(); + dfs.satisfyStoragePolicy(new Path(FILE)); + // Wait till namenode notified about the block location details + DFSTestUtil.waitExpectedStorageType(FILE, StorageType.ARCHIVE, 3, 35000, + dfs); + } + + @Test(timeout = 300000) + public void testWhenStoragePolicySetToALLSSD() + throws Exception { + try { + createCluster(); + // Change policy to ALL_SSD + dfs.setStoragePolicy(new Path(FILE), "ALL_SSD"); + + StorageType[][] newtypes = + new StorageType[][]{{StorageType.SSD, StorageType.DISK}, + {StorageType.SSD, StorageType.DISK}, + {StorageType.SSD, StorageType.DISK}}; + + // Making sure SDD based nodes added to cluster. Adding SSD based + // datanodes. + startAdditionalDNs(config, 3, NUM_OF_DATANODES, newtypes, + STORAGES_PER_DATANODE, CAPACITY, hdfsCluster); + dfs.satisfyStoragePolicy(new Path(FILE)); + hdfsCluster.triggerHeartbeats(); + // Wait till StorgePolicySatisfier Identified that block to move to SSD + // areas + DFSTestUtil.waitExpectedStorageType(FILE, StorageType.SSD, 3, 30000, dfs); + } finally { + shutdownCluster(); + } + } + + @Test(timeout = 300000) + public void testWhenStoragePolicySetToONESSD() + throws Exception { + try { + createCluster(); + // Change policy to ONE_SSD + dfs.setStoragePolicy(new Path(FILE), ONE_SSD); + + StorageType[][] newtypes = + new StorageType[][]{{StorageType.SSD, StorageType.DISK}}; + + // Making sure SDD based nodes added to cluster. Adding SSD based + // datanodes. + startAdditionalDNs(config, 1, NUM_OF_DATANODES, newtypes, + STORAGES_PER_DATANODE, CAPACITY, hdfsCluster); + dfs.satisfyStoragePolicy(new Path(FILE)); + hdfsCluster.triggerHeartbeats(); + // Wait till StorgePolicySatisfier Identified that block to move to SSD + // areas + DFSTestUtil.waitExpectedStorageType(FILE, StorageType.SSD, 1, 30000, dfs); + DFSTestUtil.waitExpectedStorageType(FILE, StorageType.DISK, 2, 30000, + dfs); + } finally { + shutdownCluster(); + } + } + + /** + * Tests to verify that the block storage movement report will be propagated + * to Namenode via datanode heartbeat. + */ + @Test(timeout = 300000) + public void testBlksStorageMovementAttemptFinishedReport() throws Exception { + try { + createCluster(); + // Change policy to ONE_SSD + dfs.setStoragePolicy(new Path(FILE), ONE_SSD); + + StorageType[][] newtypes = + new StorageType[][]{{StorageType.SSD, StorageType.DISK}}; + + // Making sure SDD based nodes added to cluster. Adding SSD based + // datanodes. + startAdditionalDNs(config, 1, NUM_OF_DATANODES, newtypes, + STORAGES_PER_DATANODE, CAPACITY, hdfsCluster); + dfs.satisfyStoragePolicy(new Path(FILE)); + hdfsCluster.triggerHeartbeats(); + + // Wait till the block is moved to SSD areas + DFSTestUtil.waitExpectedStorageType(FILE, StorageType.SSD, 1, 30000, dfs); + DFSTestUtil.waitExpectedStorageType(FILE, StorageType.DISK, 2, 30000, + dfs); + + waitForBlocksMovementAttemptReport(1, 30000); + } finally { + shutdownCluster(); + } + } + + /** + * Tests to verify that multiple files are giving to satisfy storage policy + * and should work well altogether. + */ + @Test(timeout = 300000) + public void testMultipleFilesForSatisfyStoragePolicy() throws Exception { + try { + createCluster(); + List<String> files = new ArrayList<>(); + files.add(FILE); + + // Creates 4 more files. Send all of them for satisfying the storage + // policy together. + for (int i = 0; i < 4; i++) { + String file1 = "/testMoveWhenStoragePolicyNotSatisfying_" + i; + files.add(file1); + writeContent(file1); + } + // Change policy to ONE_SSD + for (String fileName : files) { + dfs.setStoragePolicy(new Path(fileName), ONE_SSD); + dfs.satisfyStoragePolicy(new Path(fileName)); + } + + StorageType[][] newtypes = + new StorageType[][]{{StorageType.SSD, StorageType.DISK}}; + + // Making sure SDD based nodes added to cluster. Adding SSD based + // datanodes. + startAdditionalDNs(config, 1, NUM_OF_DATANODES, newtypes, + STORAGES_PER_DATANODE, CAPACITY, hdfsCluster); + hdfsCluster.triggerHeartbeats(); + + for (String fileName : files) { + // Wait till the block is moved to SSD areas + DFSTestUtil.waitExpectedStorageType( + fileName, StorageType.SSD, 1, 30000, dfs); + DFSTestUtil.waitExpectedStorageType( + fileName, StorageType.DISK, 2, 30000, dfs); + } + + waitForBlocksMovementAttemptReport(files.size(), 30000); + } finally { + shutdownCluster(); + } + } + + /** + * Tests to verify hdfsAdmin.satisfyStoragePolicy works well for file. + * @throws Exception + */ + @Test(timeout = 300000) + public void testSatisfyFileWithHdfsAdmin() throws Exception { + try { + createCluster(); + HdfsAdmin hdfsAdmin = + new HdfsAdmin(FileSystem.getDefaultUri(config), config); + // Change policy to COLD + dfs.setStoragePolicy(new Path(FILE), COLD); + + StorageType[][] newtypes = + new StorageType[][]{{StorageType.DISK, StorageType.ARCHIVE}, + {StorageType.DISK, StorageType.ARCHIVE}, + {StorageType.DISK, StorageType.ARCHIVE}}; + startAdditionalDNs(config, 3, NUM_OF_DATANODES, newtypes, + STORAGES_PER_DATANODE, CAPACITY, hdfsCluster); + + hdfsAdmin.satisfyStoragePolicy(new Path(FILE)); + + hdfsCluster.triggerHeartbeats(); + // Wait till namenode notified about the block location details + DFSTestUtil.waitExpectedStorageType(FILE, StorageType.ARCHIVE, 3, 30000, + dfs); + } finally { + shutdownCluster(); + } + } + + /** + * Tests to verify hdfsAdmin.satisfyStoragePolicy works well for dir. + * @throws Exception + */ + @Test(timeout = 300000) + public void testSatisfyDirWithHdfsAdmin() throws Exception { + try { + createCluster(); + HdfsAdmin hdfsAdmin = + new HdfsAdmin(FileSystem.getDefaultUri(config), config); + final String subDir = "/subDir"; + final String subFile1 = subDir + "/subFile1"; + final String subDir2 = subDir + "/subDir2"; + final String subFile2 = subDir2 + "/subFile2"; + dfs.mkdirs(new Path(subDir)); + writeContent(subFile1); + dfs.mkdirs(new Path(subDir2)); + writeContent(subFile2); + + // Change policy to COLD + dfs.setStoragePolicy(new Path(subDir), ONE_SSD); + + StorageType[][] newtypes = + new StorageType[][]{{StorageType.SSD, StorageType.DISK}}; + startAdditionalDNs(config, 1, NUM_OF_DATANODES, newtypes, + STORAGES_PER_DATANODE, CAPACITY, hdfsCluster); + + hdfsAdmin.satisfyStoragePolicy(new Path(subDir)); + + hdfsCluster.triggerHeartbeats(); + + // take effect for the file in the directory. + DFSTestUtil.waitExpectedStorageType( + subFile1, StorageType.SSD, 1, 30000, dfs); + DFSTestUtil.waitExpectedStorageType( + subFile1, StorageType.DISK, 2, 30000, dfs); + + // take no effect for the sub-dir's file in the directory. + DFSTestUtil.waitExpectedStorageType( + subFile2, StorageType.SSD, 1, 30000, dfs); + DFSTestUtil.waitExpectedStorageType( + subFile2, StorageType.DISK, 2, 30000, dfs); + } finally { + shutdownCluster(); + } + } + + /** + * Tests to verify hdfsAdmin.satisfyStoragePolicy exceptions. + * @throws Exception + */ + @Test(timeout = 300000) + public void testSatisfyWithExceptions() throws Exception { + try { + createCluster(); + final String nonExistingFile = "/noneExistingFile"; + hdfsCluster.getConfiguration(0). + setBoolean(DFSConfigKeys.DFS_STORAGE_POLICY_ENABLED_KEY, false); + restartNamenode(); + HdfsAdmin hdfsAdmin = + new HdfsAdmin(FileSystem.getDefaultUri(config), config); + + try { + hdfsAdmin.satisfyStoragePolicy(new Path(FILE)); + Assert.fail(String.format( + "Should failed to satisfy storage policy " + + "for %s since %s is set to false.", + FILE, DFS_STORAGE_POLICY_ENABLED_KEY)); + } catch (IOException e) { + GenericTestUtils.assertExceptionContains(String.format( + "Failed to satisfy storage policy since %s is set to false.", + DFS_STORAGE_POLICY_ENABLED_KEY), e); + } + + hdfsCluster.getConfiguration(0). + setBoolean(DFSConfigKeys.DFS_STORAGE_POLICY_ENABLED_KEY, true); + restartNamenode(); + + hdfsAdmin = new HdfsAdmin(FileSystem.getDefaultUri(config), config); + try { + hdfsAdmin.satisfyStoragePolicy(new Path(nonExistingFile)); + Assert.fail("Should throw FileNotFoundException for " + + nonExistingFile); + } catch (FileNotFoundException e) { + + } + + try { + hdfsAdmin.satisfyStoragePolicy(new Path(FILE)); + hdfsAdmin.satisfyStoragePolicy(new Path(FILE)); + } catch (Exception e) { + Assert.fail(String.format("Allow to invoke mutlipe times " + + "#satisfyStoragePolicy() api for a path %s , internally just " + + "skipping addtion to satisfy movement queue.", FILE)); + } + } finally { + shutdownCluster(); + } + } + + /** + * Tests to verify that for the given path, some of the blocks or block src + * locations(src nodes) under the given path will be scheduled for block + * movement. + * + * For example, there are two block for a file: + * + * File1 => blk_1[locations=A(DISK),B(DISK),C(DISK)], + * blk_2[locations=A(DISK),B(DISK),C(DISK)]. Now, set storage policy to COLD. + * Only one datanode is available with storage type ARCHIVE, say D. + * + * SPS will schedule block movement to the coordinator node with the details, + * blk_1[move A(DISK) -> D(ARCHIVE)], blk_2[move A(DISK) -> D(ARCHIVE)]. + */ + @Test(timeout = 300000) + public void testWhenOnlyFewTargetDatanodeAreAvailableToSatisfyStoragePolicy() + throws Exception { + try { + createCluster(); + // Change policy to COLD + dfs.setStoragePolicy(new Path(FILE), COLD); + + StorageType[][] newtypes = + new StorageType[][]{{StorageType.ARCHIVE, StorageType.ARCHIVE}}; + + // Adding ARCHIVE based datanodes. + startAdditionalDNs(config, 1, NUM_OF_DATANODES, newtypes, + STORAGES_PER_DATANODE, CAPACITY, hdfsCluster); + + dfs.satisfyStoragePolicy(new Path(FILE)); + hdfsCluster.triggerHeartbeats(); + // Wait till StorgePolicySatisfier identified that block to move to + // ARCHIVE area. + DFSTestUtil.waitExpectedStorageType(FILE, StorageType.ARCHIVE, 1, 30000, + dfs); + DFSTestUtil.waitExpectedStorageType(FILE, StorageType.DISK, 2, 30000, + dfs); + + waitForBlocksMovementAttemptReport(1, 30000); + } finally { + shutdownCluster(); + } + } + + /** + * Tests to verify that for the given path, no blocks or block src + * locations(src nodes) under the given path will be scheduled for block + * movement as there are no available datanode with required storage type. + * + * For example, there are two block for a file: + * + * File1 => blk_1[locations=A(DISK),B(DISK),C(DISK)], + * blk_2[locations=A(DISK),B(DISK),C(DISK)]. Now, set storage policy to COLD. + * No datanode is available with storage type ARCHIVE. + * + * SPS won't schedule any block movement for this path. + */ + @Test(timeout = 300000) + public void testWhenNoTargetDatanodeToSatisfyStoragePolicy() + throws Exception { + try { + createCluster(); + // Change policy to COLD + dfs.setStoragePolicy(new Path(FILE), COLD); + + StorageType[][] newtypes = + new StorageType[][]{{StorageType.DISK, StorageType.DISK}}; + // Adding DISK based datanodes + startAdditionalDNs(config, 1, NUM_OF_DATANODES, newtypes, + STORAGES_PER_DATANODE, CAPACITY, hdfsCluster); + + dfs.satisfyStoragePolicy(new Path(FILE)); + hdfsCluster.triggerHeartbeats(); + + // No block movement will be scheduled as there is no target node + // available with the required storage type. + waitForAttemptedItems(1, 30000); + DFSTestUtil.waitExpectedStorageType(FILE, StorageType.DISK, 3, 30000, + dfs); + // Since there is no target node the item will get timed out and then + // re-attempted. + waitForAttemptedItems(1, 30000); + } finally { + shutdownCluster(); + } + } + + /** + * Test to verify that satisfy worker can't move blocks. If the given block is + * pinned it shouldn't be considered for retries. + */ + @Test(timeout = 120000) + public void testMoveWithBlockPinning() throws Exception { + try{ + config.setBoolean(DFSConfigKeys.DFS_DATANODE_BLOCK_PINNING_ENABLED, true); + hdfsCluster = startCluster(config, allDiskTypes, 3, 2, CAPACITY); + + hdfsCluster.waitActive(); + dfs = hdfsCluster.getFileSystem(); + + // create a file with replication factor 3 and mark 2 pinned block + // locations. + final String file1 = createFileAndSimulateFavoredNodes(2); + + // Change policy to COLD + dfs.setStoragePolicy(new Path(file1), COLD); + + StorageType[][] newtypes = + new StorageType[][]{{StorageType.ARCHIVE, StorageType.ARCHIVE}, + {StorageType.ARCHIVE, StorageType.ARCHIVE}, + {StorageType.ARCHIVE, StorageType.ARCHIVE}}; + // Adding DISK based datanodes + startAdditionalDNs(config, 3, NUM_OF_DATANODES, newtypes, + STORAGES_PER_DATANODE, CAPACITY, hdfsCluster); + + dfs.satisfyStoragePolicy(new Path(file1)); + hdfsCluster.triggerHeartbeats(); + + // No block movement will be scheduled as there is no target node + // available with the required storage type. + waitForAttemptedItems(1, 30000); + waitForBlocksMovementAttemptReport(1, 30000); + DFSTestUtil.waitExpectedStorageType( + file1, StorageType.ARCHIVE, 1, 30000, dfs); + DFSTestUtil.waitExpectedStorageType( + file1, StorageType.DISK, 2, 30000, dfs); + } finally { + shutdownCluster(); + } + } + + /** + * Tests to verify that for the given path, only few of the blocks or block + * src locations(src nodes) under the given path will be scheduled for block + * movement. + * + * For example, there are two block for a file: + * + * File1 => two blocks and default storage policy(HOT). + * blk_1[locations=A(DISK),B(DISK),C(DISK),D(DISK),E(DISK)], + * blk_2[locations=A(DISK),B(DISK),C(DISK),D(DISK),E(DISK)]. + * + * Now, set storage policy to COLD. + * Only two Dns are available with expected storage type ARCHIVE, say A, E. + * + * SPS will schedule block movement to the coordinator node with the details, + * blk_1[move A(DISK) -> A(ARCHIVE), move E(DISK) -> E(ARCHIVE)], + * blk_2[move A(DISK) -> A(ARCHIVE), move E(DISK) -> E(ARCHIVE)]. + */ + @Test(timeout = 300000) + public void testWhenOnlyFewSourceNodesHaveMatchingTargetNodes() + throws Exception { + try { + int numOfDns = 5; + config.setLong("dfs.block.size", 1024); + allDiskTypes = + new StorageType[][]{{StorageType.DISK, StorageType.ARCHIVE}, + {StorageType.DISK, StorageType.DISK}, + {StorageType.DISK, StorageType.DISK}, + {StorageType.DISK, StorageType.DISK}, + {StorageType.DISK, StorageType.ARCHIVE}}; + hdfsCluster = startCluster(config, allDiskTypes, numOfDns, + STORAGES_PER_DATANODE, CAPACITY); + dfs = hdfsCluster.getFileSystem(); + writeContent(FILE, (short) 5); + + // Change policy to COLD + dfs.setStoragePolicy(new Path(FILE), COLD); + + dfs.satisfyStoragePolicy(new Path(FILE)); + hdfsCluster.triggerHeartbeats(); + // Wait till StorgePolicySatisfier identified that block to move to + // ARCHIVE area. + DFSTestUtil.waitExpectedStorageType(FILE, StorageType.ARCHIVE, 2, 30000, + dfs); + DFSTestUtil.waitExpectedStorageType(FILE, StorageType.DISK, 3, 30000, + dfs); + + waitForBlocksMovementAttemptReport(1, 30000); + } finally { + shutdownCluster(); + } + } + + /** + * Tests that moving block storage with in the same datanode. Let's say we + * have DN1[DISK,ARCHIVE], DN2[DISK, SSD], DN3[DISK,RAM_DISK] when + * storagepolicy set to ONE_SSD and request satisfyStoragePolicy, then block + * should move to DN2[SSD] successfully. + */ + @Test(timeout = 300000) + public void testBlockMoveInSameDatanodeWithONESSD() throws Exception { + StorageType[][] diskTypes = + new StorageType[][]{{StorageType.DISK, StorageType.ARCHIVE}, + {StorageType.DISK, StorageType.SSD}, + {StorageType.DISK, StorageType.RAM_DISK}}; + config.setLong("dfs.block.size", DEFAULT_BLOCK_SIZE); + try { + hdfsCluster = startCluster(config, diskTypes, NUM_OF_DATANODES, + STORAGES_PER_DATANODE, CAPACITY); + dfs = hdfsCluster.getFileSystem(); + writeContent(FILE); + + // Change policy to ONE_SSD + dfs.setStoragePolicy(new Path(FILE), ONE_SSD); + + dfs.satisfyStoragePolicy(new Path(FILE)); + hdfsCluster.triggerHeartbeats(); + DFSTestUtil.waitExpectedStorageType(FILE, StorageType.SSD, 1, 30000, dfs); + DFSTestUtil.waitExpectedStorageType(FILE, StorageType.DISK, 2, 30000, + dfs); + + } finally { + shutdownCluster(); + } + } + + /** + * Tests that moving block storage with in the same datanode and remote node. + * Let's say we have DN1[DISK,ARCHIVE], DN2[ARCHIVE, SSD], DN3[DISK,DISK], + * DN4[DISK,DISK] when storagepolicy set to WARM and request + * satisfyStoragePolicy, then block should move to DN1[ARCHIVE] and + * DN2[ARCHIVE] successfully. + */ + @Test(timeout = 300000) + public void testBlockMoveInSameAndRemoteDatanodesWithWARM() throws Exception { + StorageType[][] diskTypes = + new StorageType[][]{{StorageType.DISK, StorageType.ARCHIVE}, + {StorageType.ARCHIVE, StorageType.SSD}, + {StorageType.DISK, StorageType.DISK}, + {StorageType.DISK, StorageType.DISK}}; + + config.setLong("dfs.block.size", DEFAULT_BLOCK_SIZE); + try { + hdfsCluster = startCluster(config, diskTypes, diskTypes.length, + STORAGES_PER_DATANODE, CAPACITY); + dfs = hdfsCluster.getFileSystem(); + writeContent(FILE); + + // Change policy to WARM + dfs.setStoragePolicy(new Path(FILE), "WARM"); + dfs.satisfyStoragePolicy(new Path(FILE)); + hdfsCluster.triggerHeartbeats(); + + DFSTestUtil.waitExpectedStorageType(FILE, StorageType.DISK, 1, 30000, + dfs); + DFSTestUtil.waitExpectedStorageType(FILE, StorageType.ARCHIVE, 2, 30000, + dfs); + } finally { + shutdownCluster(); + } + } + + /** + * If replica with expected storage type already exist in source DN then that + * DN should be skipped. + */ + @Test(timeout = 300000) + public void testSPSWhenReplicaWithExpectedStorageAlreadyAvailableInSource() + throws Exception { + StorageType[][] diskTypes = new StorageType[][] { + {StorageType.DISK, StorageType.ARCHIVE}, + {StorageType.DISK, StorageType.ARCHIVE}, + {StorageType.DISK, StorageType.ARCHIVE}}; + + try { + hdfsCluster = startCluster(config, diskTypes, diskTypes.length, + STORAGES_PER_DATANODE, CAPACITY); + dfs = hdfsCluster.getFileSystem(); + // 1. Write two replica on disk + DFSTestUtil.createFile(dfs, new Path(FILE), DEFAULT_BLOCK_SIZE, + (short) 2, 0); + // 2. Change policy to COLD, so third replica will be written to ARCHIVE. + dfs.setStoragePolicy(new Path(FILE), "COLD"); + + // 3.Change replication factor to 3. + dfs.setReplication(new Path(FILE), (short) 3); + + DFSTestUtil.waitExpectedStorageType(FILE, StorageType.DISK, 2, 30000, + dfs); + DFSTestUtil.waitExpectedStorageType(FILE, StorageType.ARCHIVE, 1, 30000, + dfs); + + // 4. Change policy to HOT, so we can move the all block to DISK. + dfs.setStoragePolicy(new Path(FILE), "HOT"); + + // 4. Satisfy the policy. + dfs.satisfyStoragePolicy(new Path(FILE)); + + // 5. Block should move successfully . + DFSTestUtil.waitExpectedStorageType(FILE, StorageType.DISK, 3, 30000, + dfs); + } finally { + shutdownCluster(); + } + } + + /** + * Tests that movements should not be assigned when there is no space in + * target DN. + */ + @Test(timeout = 300000) + public void testChooseInSameDatanodeWithONESSDShouldNotChooseIfNoSpace() + throws Exception { + StorageType[][] diskTypes = + new StorageType[][]{{StorageType.DISK, StorageType.DISK}, + {StorageType.DISK, StorageType.SSD}, + {StorageType.DISK, StorageType.DISK}}; + config.setLong("dfs.block.size", 2 * DEFAULT_BLOCK_SIZE); + long dnCapacity = 1024 * DEFAULT_BLOCK_SIZE + (2 * DEFAULT_BLOCK_SIZE - 1); + try { + hdfsCluster = startCluster(config, diskTypes, NUM_OF_DATANODES, + STORAGES_PER_DATANODE, dnCapacity); + dfs = hdfsCluster.getFileSystem(); + writeContent(FILE); + + // Change policy to ONE_SSD + dfs.setStoragePolicy(new Path(FILE), ONE_SSD); + Path filePath = new Path("/testChooseInSameDatanode"); + final FSDataOutputStream out = + dfs.create(filePath, false, 100, (short) 1, 2 * DEFAULT_BLOCK_SIZE); + try { + dfs.setStoragePolicy(filePath, ONE_SSD); + // Try to fill up SSD part by writing content + long remaining = dfs.getStatus().getRemaining() / (3 * 2); + for (int i = 0; i < remaining; i++) { + out.write(i); + } + } finally { + out.close(); + } + hdfsCluster.triggerHeartbeats(); + ArrayList<DataNode> dataNodes = hdfsCluster.getDataNodes(); + // Temporarily disable heart beats, so that we can assert whether any + // items schedules for DNs even though DN's does not have space to write. + // Disabling heart beats can keep scheduled items on DatanodeDescriptor + // itself. + for (DataNode dataNode : dataNodes) { + DataNodeTestUtils.setHeartbeatsDisabledForTests(dataNode, true); + } + dfs.satisfyStoragePolicy(new Path(FILE)); + + // Wait for items to be processed + waitForAttemptedItems(1, 30000); + + // Enable heart beats now + for (DataNode dataNode : dataNodes) { + DataNodeTestUtils.setHeartbeatsDisabledForTests(dataNode, false); + } + hdfsCluster.triggerHeartbeats(); + + DFSTestUtil.waitExpectedStorageType(FILE, StorageType.DISK, 3, 30000, + dfs); + DFSTestUtil.waitExpectedStorageType(FILE, StorageType.SSD, 0, 30000, dfs); + } finally { + shutdownCluster(); + } + } + + /** + * Tests that Xattrs should be cleaned if satisfy storage policy called on EC + * file with unsuitable storage policy set. + * + * @throws Exception + */ + @Test(timeout = 300000) + public void testSPSShouldNotLeakXattrIfSatisfyStoragePolicyCallOnECFiles() + throws Exception { + StorageType[][] diskTypes = + new StorageType[][]{{StorageType.SSD, StorageType.DISK}, + {StorageType.SSD, StorageType.DISK}, + {StorageType.SSD, StorageType.DISK}, + {StorageType.SSD, StorageType.DISK}, + {StorageType.SSD, StorageType.DISK}, + {StorageType.DISK, StorageType.SSD}, + {StorageType.DISK, StorageType.SSD}, + {StorageType.DISK, StorageType.SSD}, + {StorageType.DISK, StorageType.SSD}, + {StorageType.DISK, StorageType.SSD}}; + + int defaultStripedBlockSize = + StripedFileTestUtil.getDefaultECPolicy().getCellSize() * 4; + config.setLong(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, defaultStripedBlockSize); + config.setLong(DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_KEY, 1L); + config.setLong(DFSConfigKeys.DFS_NAMENODE_REDUNDANCY_INTERVAL_SECONDS_KEY, + 1L); + config.setBoolean(DFSConfigKeys.DFS_NAMENODE_REDUNDANCY_CONSIDERLOAD_KEY, + false); + try { + hdfsCluster = startCluster(config, diskTypes, diskTypes.length, + STORAGES_PER_DATANODE, CAPACITY); + dfs = hdfsCluster.getFileSystem(); + dfs.enableErasureCodingPolicy( + StripedFileTestUtil.getDefaultECPolicy().getName()); + + // set "/foo" directory with ONE_SSD storage policy. + ClientProtocol client = NameNodeProxies.createProxy(config, + hdfsCluster.getFileSystem(0).getUri(), ClientProtocol.class) + .getProxy(); + String fooDir = "/foo"; + client.mkdirs(fooDir, new FsPermission((short) 777), true); + // set an EC policy on "/foo" directory + client.setErasureCodingPolicy(fooDir, + StripedFileTestUtil.getDefaultECPolicy().getName()); + + // write file to fooDir + final String testFile = "/foo/bar"; + long fileLen = 20 * defaultStripedBlockSize; + DFSTestUtil.createFile(dfs, new Path(testFile), fileLen, (short) 3, 0); + + // ONESSD is unsuitable storage policy on EC files + client.setStoragePolicy(fooDir, HdfsConstants.ONESSD_STORAGE_POLICY_NAME); + dfs.satisfyStoragePolicy(new Path(testFile)); + + // Thread.sleep(9000); // To make sure SPS triggered + // verify storage types and locations + LocatedBlocks locatedBlocks = + client.getBlockLocations(testFile, 0, fileLen); + for (LocatedBlock lb : locatedBlocks.getLocatedBlocks()) { + for (StorageType type : lb.getStorageTypes()) { + Assert.assertEquals(StorageType.DISK, type); + } + } + + // Make sure satisfy xattr has been removed. + DFSTestUtil.waitForXattrRemoved(testFile, XATTR_SATISFY_STORAGE_POLICY, + hdfsCluster.getNamesystem(), 30000); + } finally { + shutdownCluster(); + } + } + + /** + * Test SPS with empty file. + * 1. Create one empty file. + * 2. Call satisfyStoragePolicy for empty file. + * 3. SPS should skip this file and xattr should not be added for empty file. + */ + @Test(timeout = 300000) + public void testSPSWhenFileLengthIsZero() throws Exception { + try { + hdfsCluster = startCluster(config, allDiskTypes, NUM_OF_DATANODES, + STORAGES_PER_DATANODE, CAPACITY); + hdfsCluster.waitActive(); + DistributedFileSystem fs = hdfsCluster.getFileSystem(); + Path filePath = new Path("/zeroSizeFile"); + DFSTestUtil.createFile(fs, filePath, 0, (short) 1, 0); + FSEditLog editlog = hdfsCluster.getNameNode().getNamesystem() + .getEditLog(); + long lastWrittenTxId = editlog.getLastWrittenTxId(); + fs.satisfyStoragePolicy(filePath); + Assert.assertEquals("Xattr should not be added for the file", + lastWrittenTxId, editlog.getLastWrittenTxId()); + INode inode = hdfsCluster.getNameNode().getNamesystem().getFSDirectory() + .getINode(filePath.toString()); + Assert.assertTrue("XAttrFeature should be null for file", + inode.getXAttrFeature() == null); + } finally { + shutdownCluster(); + } + } + + /** + * Test SPS for low redundant file blocks. + * 1. Create cluster with 3 datanode. + * 1. Create one file with 3 replica. + * 2. Set policy and call satisfyStoragePolicy for file. + * 3. Stop NameNode and Datanodes. + * 4. Start NameNode with 2 datanode and wait for block movement. + * 5. Start third datanode. + * 6. Third Datanode replica also should be moved in proper + * sorage based on policy. + */ + @Test(timeout = 300000) + public void testSPSWhenFileHasLowRedundancyBlocks() throws Exception { + try { + config.set(DFSConfigKeys + .DFS_STORAGE_POLICY_SATISFIER_RECHECK_TIMEOUT_MILLIS_KEY, + "3000"); + config.set(DFSConfigKeys + .DFS_STORAGE_POLICY_SATISFIER_SELF_RETRY_TIMEOUT_MILLIS_KEY, + "5000"); + StorageType[][] newtypes = new StorageType[][] { + {StorageType.ARCHIVE, StorageType.DISK}, + {StorageType.ARCHIVE, StorageType.DISK}, + {StorageType.ARCHIVE, StorageType.DISK}}; + hdfsCluster = startCluster(config, newtypes, 3, 2, CAPACITY); + hdfsCluster.waitActive(); + DistributedFileSystem fs = hdfsCluster.getFileSystem(); + Path filePath = new Path("/zeroSizeFile"); + DFSTestUtil.createFile(fs, filePath, 1024, (short) 3, 0); + fs.setStoragePolicy(filePath, "COLD"); + List<DataNodeProperties> list = new ArrayList<>(); + list.add(hdfsCluster.stopDataNode(0)); + list.add(hdfsCluster.stopDataNode(0)); + list.add(hdfsCluster.stopDataNode(0)); + restartNamenode(); + hdfsCluster.restartDataNode(list.get(0), false); + hdfsCluster.restartDataNode(list.get(1), false); + hdfsCluster.waitActive(); + fs.satisfyStoragePolicy(filePath); + DFSTestUtil.waitExpectedStorageType(filePath.toString(), + StorageType.ARCHIVE, 2, 30000, hdfsCluster.getFileSystem()); + hdfsCluster.restartDataNode(list.get(2), false); + DFSTestUtil.waitExpectedStorageType(filePath.toString(), + StorageType.ARCHIVE, 3, 30000, hdfsCluster.getFileSystem()); + } finally { + shutdownCluster(); + } + } + + /** + * Test SPS for extra redundant file blocks. + * 1. Create cluster with 5 datanode. + * 2. Create one file with 5 replica. + * 3. Set file replication to 3. + * 4. Set policy and call satisfyStoragePolicy for file. + * 5. Block should be moved successfully. + */ + @Test(timeout = 300000) + public void testSPSWhenFileHasExcessRedundancyBlocks() throws Exception { + try { + config.set(DFSConfigKeys + .DFS_STORAGE_POLICY_SATISFIER_RECHECK_TIMEOUT_MILLIS_KEY, + "3000"); + config.set(DFSConfigKeys + .DFS_STORAGE_POLICY_SATISFIER_SELF_RETRY_TIMEOUT_MILLIS_KEY, + "5000"); + StorageType[][] newtypes = new StorageType[][] { + {StorageType.ARCHIVE, StorageType.DISK}, + {StorageType.ARCHIVE, StorageType.DISK}, + {StorageType.ARCHIVE, StorageType.DISK}, + {StorageType.ARCHIVE, StorageType.DISK}, + {StorageType.ARCHIVE, StorageType.DISK}}; + hdfsCluster = startCluster(config, newtypes, 5, 2, CAPACITY); + hdfsCluster.waitActive(); + DistributedFileSystem fs = hdfsCluster.getFileSystem(); + Path filePath = new Path("/zeroSizeFile"); + DFSTestUtil.createFile(fs, filePath, 1024, (short) 5, 0); + fs.setReplication(filePath, (short) 3); + LogCapturer logs = GenericTestUtils.LogCapturer.captureLogs( + LogFactory.getLog(BlockStorageMovementAttemptedItems.class)); + fs.setStoragePolicy(filePath, "COLD"); + fs.satisfyStoragePolicy(filePath); + DFSTestUtil.waitExpectedStorageType(filePath.toString(), + StorageType.ARCHIVE, 3, 60000, hdfsCluster.getFileSystem()); + assertFalse("Log output does not contain expected log message: ", + logs.getOutput().contains("some of the blocks are low redundant")); + } finally { + shutdownCluster(); + } + } + + /** + * Test SPS for empty directory, xAttr should be removed. + */ + @Test(timeout = 300000) + public void testSPSForEmptyDirectory() throws IOException, TimeoutException, + InterruptedException { + try { + hdfsCluster = startCluster(config, allDiskTypes, NUM_OF_DATANODES, + STORAGES_PER_DATANODE, CAPACITY); + hdfsCluster.waitActive(); + DistributedFileSystem fs = hdfsCluster.getFileSystem(); + Path emptyDir = new Path("/emptyDir"); + fs.mkdirs(emptyDir); + fs.satisfyStoragePolicy(emptyDir); + // Make sure satisfy xattr has been removed. + DFSTestUtil.waitForXattrRemoved("/emptyDir", + XATTR_SATISFY_STORAGE_POLICY, hdfsCluster.getNamesystem(), 30000); + } finally { + shutdownCluster(); + } + } + + /** + * Test SPS for not exist directory. + */ + @Test(timeout = 300000) + public void testSPSForNonExistDirectory() throws Exception { + try { + hdfsCluster = startCluster(config, allDiskTypes, NUM_OF_DATANODES, + STORAGES_PER_DATANODE, CAPACITY); + hdfsCluster.waitActive(); + DistributedFileSystem fs = hdfsCluster.getFileSystem(); + Path emptyDir = new Path("/emptyDir"); + try { + fs.satisfyStoragePolicy(emptyDir); + fail("FileNotFoundException should throw"); + } catch (FileNotFoundException e) { + // nothing to do + } + } finally { + shutdownCluster(); + } + } + + /** + * Test SPS for directory tree which doesn't have files. + */ + @Test(timeout = 300000) + public void testSPSWithDirectoryTreeWithoutFile() throws Exception { + try { + hdfsCluster = startCluster(config, allDiskTypes, NUM_OF_DATANODES, + STORAGES_PER_DATANODE, CAPACITY); + hdfsCluster.waitActive(); + // Create directories + /* + * root + * | + * A--------C--------D + * | + * G----H----I + * | + * O + */ + DistributedFileSystem fs = hdfsCluster.getFileSystem(); + fs.mkdirs(new Path("/root/C/H/O")); + fs.mkdirs(new Path("/root/A")); + fs.mkdirs(new Path("/root/D")); + fs.mkdirs(new Path("/root/C/G")); + fs.mkdirs(new Path("/root/C/I")); + fs.satisfyStoragePolicy(new Path("/root")); + // Make sure satisfy xattr has been removed. + DFSTestUtil.waitForXattrRemoved("/root", + XATTR_SATISFY_STORAGE_POLICY, hdfsCluster.getNamesystem(), 30000); + } finally { + shutdownCluster(); + } + } + + /** + * Test SPS for directory which has multilevel directories. + */ + @Test(timeout = 300000) + public void testMultipleLevelDirectoryForSatisfyStoragePolicy() + throws Exception { + try { + StorageType[][] diskTypes = new StorageType[][] { + {StorageType.DISK, StorageType.ARCHIVE}, + {StorageType.ARCHIVE, StorageType.SSD}, + {StorageType.DISK, StorageType.DISK}}; + config.setLong("dfs.block.size", DEFAULT_BLOCK_SIZE); + hdfsCluster = startCluster(config, diskTypes, diskTypes.length, + STORAGES_PER_DATANODE, CAPACITY); + dfs = hdfsCluster.getFileSystem(); + createDirectoryTree(dfs); + + List<String> files = getDFSListOfTree(); + dfs.setStoragePolicy(new Path("/root"), COLD); + dfs.satisfyStoragePolicy(new Path("/root")); + for (String fileName : files) { + // Wait till the block is moved to ARCHIVE + DFSTestUtil.waitExpectedStorageType(fileName, StorageType.ARCHIVE, 2, + 30000, dfs); + } + } finally { + shutdownCluster(); + } + } + + /** + * Test storage move blocks while under replication block tasks exists in the + * system. So, both will share the max transfer streams. + * + * 1. Create cluster with 3 datanode. + * 2. Create 20 files with 2 replica. + * 3. Start 2 more DNs with DISK & SSD types + * 4. SetReplication factor for the 1st 10 files to 4 to trigger replica task + * 5. Set policy to SSD to the 2nd set of files from 11-20 + * 6. Call SPS for 11-20 files to trigger move block tasks to new DNs + * 7. Wait for the under replica and SPS tasks completion + */ + @Test(timeout = 300000) + public void testMoveBlocksWithUnderReplicatedBlocks() throws Exception { + try { + config.setInt(DFSConfigKeys.DFS_NAMENODE_REPLICATION_MAX_STREAMS_KEY, 3); + config.setLong("dfs.block.size", DEFAULT_BLOCK_SIZE); + config.set(DFSConfigKeys + .DFS_STORAGE_POLICY_SATISFIER_RECHECK_TIMEOUT_MILLIS_KEY, + "3000"); + config.set(DFSConfigKeys + .DFS_STORAGE_POLICY_SATISFIER_SELF_RETRY_TIMEOUT_MILLIS_KEY, + "5000"); + + StorageType[][] storagetypes = new StorageType[][] { + {StorageType.ARCHIVE, StorageType.DISK}, + {StorageType.ARCHIVE, StorageType.DISK}}; + + hdfsCluster = startCluster(config, storagetypes, 2, 2, CAPACITY); + hdfsCluster.waitActive(); + dfs = hdfsCluster.getFileSystem(); + + // Below files will be used for pending replication block tasks. + for (int i=1; i<=20; i++){ + Path filePath = new Path("/file" + i); + DFSTestUtil.createFile(dfs, filePath, DEFAULT_BLOCK_SIZE * 5, (short) 2, + 0); + } + + StorageType[][] newtypes = + new StorageType[][]{{StorageType.DISK, StorageType.SSD}, + {StorageType.DISK, StorageType.SSD}}; + startAdditionalDNs(config, 2, NUM_OF_DATANODES, newtypes, + STORAGES_PER_DATANODE, CAPACITY, hdfsCluster); + + // increase replication factor to 4 for the first 10 files and thus + // initiate replica tasks + for (int i=1; i<=10; i++){ + Path filePath = new Path("/file" + i); + dfs.setReplication(filePath, (short) 4); + } + + // invoke SPS for 11-20 files + for (int i = 11; i <= 20; i++) { + Path filePath = new Path("/file" + i); + dfs.setStoragePolicy(filePath, "ALL_SSD"); + dfs.satisfyStoragePolicy(filePath); + } + + for (int i = 1; i <= 10; i++) { + Path filePath = new Path("/file" + i); + DFSTestUtil.waitExpectedStorageType(filePath.toString(), + StorageType.DISK, 4, 60000, hdfsCluster.getFileSystem()); + } + for (int i = 11; i <= 20; i++) { + Path filePath = new Path("/file" + i); + DFSTestUtil.waitExpectedStorageType(filePath.toString(), + StorageType.SSD, 2, 30000, hdfsCluster.getFileSystem()); + } + } finally { + shutdownCluster(); + } + } + + private static void createDirectoryTree(DistributedFileSystem dfs) + throws Exception { + // tree structure + /* + * root + * | + * A--------B--------C--------D--------E + * | | + * F----G----H----I J----K----L----M + * | | + * N----O----P Q----R----S + * | | + * T U + */ + // create root Node and child + dfs.mkdirs(new Path("/root")); + DFSTestUtil.createFile(dfs, new Path("/root/A"), 1024, (short) 3, 0); + dfs.mkdirs(new Path("/root/B")); + DFSTestUtil.createFile(dfs, new Path("/root/C"), 1024, (short) 3, 0); + dfs.mkdirs(new Path("/root/D")); + DFSTestUtil.createFile(dfs, new Path("/root/E"), 1024, (short) 3, 0); + + // Create /root/B child + DFSTestUtil.createFile(dfs, new Path("/root/B/F"), 1024, (short) 3, 0); + dfs.mkdirs(new Path("/root/B/G")); + DFSTestUtil.createFile(dfs, new Path("/root/B/H"), 1024, (short) 3, 0); + DFSTestUtil.createFile(dfs, new Path("/root/B/I"), 1024, (short) 3, 0); + + // Create /root/D child + DFSTestUtil.createFile(dfs, new Path("/root/D/J"), 1024, (short) 3, 0); + DFSTestUtil.createFile(dfs, new Path("/root/D/K"), 1024, (short) 3, 0); + dfs.mkdirs(new Path("/root/D/L")); + DFSTestUtil.createFile(dfs, new Path("/root/D/M"), 1024, (short) 3, 0); + + // Create /root/B/G child + DFSTestUtil.createFile(dfs, new Path("/root/B/G/N"), 1024, (short) 3, 0); + DFSTestUtil.createFile(dfs, new Path("/root/B/G/O"), 1024, (short) 3, 0); + dfs.mkdirs(new Path("/root/B/G/P")); + + // Create /root/D/L child + dfs.mkdirs(new Path("/root/D/L/Q")); + DFSTestUtil.createFile(dfs, new Path("/root/D/L/R"), 1024, (short) 3, 0); + DFSTestUtil.createFile(dfs, new Path("/root/D/L/S"), 1024, (short) 3, 0); + + // Create /root/B/G/P child + DFSTestUtil.createFile(dfs, new Path("/root/B/G/P/T"), 1024, (short) 3, 0); + + // Create /root/D/L/Q child + DFSTestUtil.createFile(dfs, new Path("/root/D/L/Q/U"), 1024, (short) 3, 0); + } + + private List<String> getDFSListOfTree() { + List<String> dfsList = new ArrayList<>(); + dfsList.add("/root/A"); + dfsList.add("/root/B/F"); + dfsList.add("/root/B/G/N"); + dfsList.add("/root/B/G/O"); + dfsList.add("/root/B/G/P/T"); + dfsList.add("/root/B/H"); + dfsList.add("/root/B/I"); + dfsList.add("/root/C"); + dfsList.add("/root/D/J"); + dfsList.add("/root/D/K"); + dfsList.add("/root/D/L/Q/U"); + dfsList.add("/root/D/L/R"); + dfsList.add("/root/D/L/S"); + dfsList.add("/root/D/M"); + dfsList.add("/root/E"); + return dfsList; + } + + private String createFileAndSimulateFavoredNodes(int favoredNodesCount) + throws IOException { + ArrayList<DataNode> dns = hdfsCluster.getDataNodes(); + final String file1 = "/testMoveWithBlockPinning"; + // replication factor 3 + InetSocketAddress[] favoredNodes = new InetSocketAddress[favoredNodesCount]; + for (int i = 0; i < favoredNodesCount; i++) { + favoredNodes[i] = dns.get(i).getXferAddress(); + } + DFSTestUtil.createFile(dfs, new Path(file1), false, 1024, 100, + DEFAULT_BLOCK_SIZE, (short) 3, 0, false, favoredNodes); + + LocatedBlocks locatedBlocks = dfs.getClient().getLocatedBlocks(file1, 0); + Assert.assertEquals("Wrong block count", 1, + locatedBlocks.locatedBlockCount()); + + // verify storage type before movement + LocatedBlock lb = locatedBlocks.get(0); + StorageType[] storageTypes = lb.getStorageTypes(); + for (StorageType storageType : storageTypes) { + Assert.assertTrue(StorageType.DISK == storageType); + } + + // Mock FsDatasetSpi#getPinning to show that the block is pinned. + DatanodeInfo[] locations = lb.getLocations(); + Assert.assertEquals(3, locations.length); + Assert.assertTrue(favoredNodesCount < locations.length); + for(DatanodeInfo dnInfo: locations){ + LOG.info("Simulate block pinning in datanode {}", + locations[favoredNodesCount]); + DataNode dn = hdfsCluster.getDataNode(dnInfo.getIpcPort()); + InternalDataNodeTestUtils.mockDatanodeBlkPinning(dn, true); + favoredNodesCount--; + if (favoredNodesCount <= 0) { + break; // marked favoredNodesCount number of pinned block location + } + } + return file1; + } + + public void waitForAttemptedItems(long expectedBlkMovAttemptedCount, + int timeout) throws TimeoutException, InterruptedException { + GenericTestUtils.waitFor(new Supplier<Boolean>() { + @Override + public Boolean get() { + LOG.info("expectedAttemptedItemsCount={} actualAttemptedItemsCount={}", + expectedBlkMovAttemptedCount, + ((BlockStorageMovementAttemptedItems) (externalSps + .getAttemptedItemsMonitor())).getAttemptedItemsCount()); + return ((BlockStorageMovementAttemptedItems) (externalSps + .getAttemptedItemsMonitor())) + .getAttemptedItemsCount() == expectedBlkMovAttemptedCount; + } + }, 100, timeout); + } + + public void waitForBlocksMovementAttemptReport( + long expectedMovementFinishedBlocksCount, int timeout) + throws TimeoutException, InterruptedException { + GenericTestUtils.waitFor(new Supplier<Boolean>() { + @Override + public Boolean get() { + int actualCount = externalSps.getAttemptedItemsMonitor() + .getAttemptedItemsCount(); + LOG.info("MovementFinishedBlocks: expectedCount={} actualCount={}", + expectedMovementFinishedBlocksCount, actualCount); + return actualCount + >= expectedMovementFinishedBlocksCount; + } + }, 100, timeout); + } + + public void writeContent(final String fileName) throws IOException { + writeContent(fileName, (short) 3); + } + + private void writeContent(final String fileName, short replicatonFactor) + throws IOException { + // write to DISK + final FSDataOutputStream out = dfs.create(new Path(fileName), + replicatonFactor); + for (int i = 0; i < 1024; i++) { + out.write(i); + } + out.close(); + } + + private void startAdditionalDNs(final Configuration conf, + int newNodesRequired, int existingNodesNum, StorageType[][] newTypes, + int storagesPerDn, long nodeCapacity, final MiniDFSCluster cluster) + throws IOException { + long[][] capacities; + existingNodesNum += newNodesRequired; + capacities = new long[newNodesRequired][storagesPerDn]; + for (int i = 0; i < newNodesRequired; i++) { + for (int j = 0; j < storagesPerDn; j++) { + capacities[i][j] = nodeCapacity; + } + } + + cluster.startDataNodes(conf, newNodesRequired, newTypes, true, null, null, + null, capacities, null, false, false, false, null); + cluster.triggerHeartbeats(); + } + + /** + * Implementation of listener callback, where it collects all the sps move + * attempted blocks for assertion. + */ + public static final class ExternalBlockMovementListener + implements BlockMovementListener { + + private List<Block> actualBlockMovements = new ArrayList<>(); + + @Override + public void notifyMovementTriedBlocks(Block[] moveAttemptFinishedBlks) { + for (Block block : moveAttemptFinishedBlks) { + actualBlockMovements.add(block); + } + LOG.info("Movement attempted blocks:{}", actualBlockMovements); + } + + public List<Block> getActualBlockMovements() { + return actualBlockMovements; + } + + public void clear() { + actualBlockMovements.clear(); + } + } }
http://git-wip-us.apache.org/repos/asf/hadoop/blob/e5bd5714/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/tools/TestStoragePolicyCommands.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/tools/TestStoragePolicyCommands.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/tools/TestStoragePolicyCommands.java index 28838a6..ad77684 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/tools/TestStoragePolicyCommands.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/tools/TestStoragePolicyCommands.java @@ -50,7 +50,7 @@ public class TestStoragePolicyCommands { public void clusterSetUp() throws IOException, URISyntaxException { conf = new HdfsConfiguration(); conf.set(DFSConfigKeys.DFS_STORAGE_POLICY_SATISFIER_MODE_KEY, - StoragePolicySatisfierMode.INTERNAL.toString()); + StoragePolicySatisfierMode.EXTERNAL.toString()); StorageType[][] newtypes = new StorageType[][] { {StorageType.ARCHIVE, StorageType.DISK}}; cluster = new MiniDFSCluster.Builder(conf).numDataNodes(REPL) http://git-wip-us.apache.org/repos/asf/hadoop/blob/e5bd5714/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/tools/TestStoragePolicySatisfyAdminCommands.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/tools/TestStoragePolicySatisfyAdminCommands.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/tools/TestStoragePolicySatisfyAdminCommands.java index 8a62e0e..1ab7788 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/tools/TestStoragePolicySatisfyAdminCommands.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/tools/TestStoragePolicySatisfyAdminCommands.java @@ -29,6 +29,11 @@ import org.apache.hadoop.hdfs.DistributedFileSystem; import org.apache.hadoop.hdfs.HdfsConfiguration; import org.apache.hadoop.hdfs.MiniDFSCluster; import org.apache.hadoop.hdfs.protocol.HdfsConstants.StoragePolicySatisfierMode; +import org.apache.hadoop.hdfs.server.balancer.NameNodeConnector; +import org.apache.hadoop.hdfs.server.common.HdfsServerConstants; +import org.apache.hadoop.hdfs.server.namenode.sps.Context; +import org.apache.hadoop.hdfs.server.namenode.sps.StoragePolicySatisfier; +import org.apache.hadoop.hdfs.server.sps.ExternalSPSContext; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -43,12 +48,13 @@ public class TestStoragePolicySatisfyAdminCommands { private Configuration conf = null; private MiniDFSCluster cluster = null; private DistributedFileSystem dfs = null; + private StoragePolicySatisfier externalSps = null; @Before public void clusterSetUp() throws IOException, URISyntaxException { conf = new HdfsConfiguration(); conf.set(DFSConfigKeys.DFS_STORAGE_POLICY_SATISFIER_MODE_KEY, - StoragePolicySatisfierMode.INTERNAL.toString()); + StoragePolicySatisfierMode.EXTERNAL.toString()); // Reduced refresh cycle to update latest datanodes. conf.setLong(DFSConfigKeys.DFS_SPS_DATANODE_CACHE_REFRESH_INTERVAL_MS, 1000); @@ -58,6 +64,14 @@ public class TestStoragePolicySatisfyAdminCommands { .storageTypes(newtypes).build(); cluster.waitActive(); dfs = cluster.getFileSystem(); + NameNodeConnector nnc = DFSTestUtil.getNameNodeConnector(conf, + HdfsServerConstants.MOVER_ID_PATH, 1, false); + + StoragePolicySatisfier externalSps = new StoragePolicySatisfier(conf); + Context externalCtxt = new ExternalSPSContext(externalSps, nnc); + + externalSps.init(externalCtxt); + externalSps.start(true, StoragePolicySatisfierMode.EXTERNAL); } @After @@ -70,6 +84,9 @@ public class TestStoragePolicySatisfyAdminCommands { cluster.shutdown(); cluster = null; } + if (externalSps != null) { + externalSps.stopGracefully(); + } } @Test(timeout = 30000) @@ -92,41 +109,4 @@ public class TestStoragePolicySatisfyAdminCommands { DFSTestUtil.waitExpectedStorageType(file, StorageType.ARCHIVE, 1, 30000, dfs); } - - @Test(timeout = 30000) - public void testIsSatisfierRunningCommand() throws Exception { - final String file = "/testIsSatisfierRunningCommand"; - DFSTestUtil.createFile(dfs, new Path(file), SIZE, REPL, 0); - final StoragePolicyAdmin admin = new StoragePolicyAdmin(conf); - DFSTestUtil.toolRun(admin, "-isInternalSatisfierRunning", 0, "yes"); - - cluster.getNameNode().reconfigureProperty( - DFSConfigKeys.DFS_STORAGE_POLICY_SATISFIER_MODE_KEY, - StoragePolicySatisfierMode.NONE.toString()); - cluster.waitActive(); - - DFSTestUtil.toolRun(admin, "-isInternalSatisfierRunning", 0, "no"); - - // Test with unnecessary args - DFSTestUtil.toolRun(admin, "-isInternalSatisfierRunning status", 1, - "Can't understand arguments: "); - } - - @Test(timeout = 90000) - public void testSatisfyStoragePolicyCommandWithWaitOption() - throws Exception { - final String file = "/testSatisfyStoragePolicyCommandWithWaitOption"; - DFSTestUtil.createFile(dfs, new Path(file), SIZE, REPL, 0); - - final StoragePolicyAdmin admin = new StoragePolicyAdmin(conf); - - DFSTestUtil.toolRun(admin, "-setStoragePolicy -path " + file - + " -policy COLD", 0, "Set storage policy COLD on " + file.toString()); - - DFSTestUtil.toolRun(admin, "-satisfyStoragePolicy -w -path " + file, 0, - "Waiting for satisfy the policy"); - - DFSTestUtil.waitExpectedStorageType(file, StorageType.ARCHIVE, 1, 30000, - dfs); - } } --------------------------------------------------------------------- To unsubscribe, e-mail: common-commits-unsubscr...@hadoop.apache.org For additional commands, e-mail: common-commits-h...@hadoop.apache.org