TestPersistentStoragePolicySatisfier.java
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
* <p>
* http://www.apache.org/licenses/LICENSE-2.0
* <p>
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.hadoop.hdfs.server.namenode;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.fs.XAttr;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.DFSTestUtil;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.HdfsConfiguration;
import org.apache.hadoop.hdfs.MiniDFSCluster;
import org.apache.hadoop.hdfs.MiniDFSCluster.DataNodeProperties;
import org.apache.hadoop.hdfs.protocol.HdfsConstants.StoragePolicySatisfierMode;
import org.apache.hadoop.hdfs.MiniDFSNNTopology;
import org.apache.hadoop.hdfs.server.balancer.NameNodeConnector;
import org.apache.hadoop.hdfs.server.common.HdfsServerConstants;
import org.apache.hadoop.hdfs.server.namenode.ha.HATestUtil;
import org.apache.hadoop.hdfs.server.namenode.sps.StoragePolicySatisfier;
import org.apache.hadoop.hdfs.server.sps.ExternalSPSContext;
import org.apache.hadoop.test.GenericTestUtils;
import org.junit.Test;
import java.util.function.Supplier;
import java.io.IOException;
import java.util.List;
import static org.apache.hadoop.hdfs.server.common.HdfsServerConstants.XATTR_SATISFY_STORAGE_POLICY;
import static org.junit.Assert.*;
/**
* Test persistence of satisfying files/directories.
*/
public class TestPersistentStoragePolicySatisfier {
private static Configuration conf;
private static MiniDFSCluster cluster;
private static DistributedFileSystem fs;
private NameNodeConnector nnc;
private StoragePolicySatisfier sps;
private ExternalSPSContext ctxt;
private static Path testFile =
new Path("/testFile");
private static String testFileName = testFile.toString();
private static Path parentDir = new Path("/parentDir");
private static Path parentFile = new Path(parentDir, "parentFile");
private static String parentFileName = parentFile.toString();
private static Path childDir = new Path(parentDir, "childDir");
private static Path childFile = new Path(childDir, "childFile");
private static String childFileName = childFile.toString();
private static final String COLD = "COLD";
private static final String WARM = "WARM";
private static final String ONE_SSD = "ONE_SSD";
private static StorageType[][] storageTypes = new StorageType[][] {
{StorageType.DISK, StorageType.ARCHIVE, StorageType.SSD},
{StorageType.DISK, StorageType.ARCHIVE, StorageType.SSD},
{StorageType.DISK, StorageType.ARCHIVE, StorageType.SSD}
};
private final int timeout = 90000;
/**
* Setup environment for every test case.
* @throws IOException
*/
public void clusterSetUp() throws Exception {
clusterSetUp(false, new HdfsConfiguration());
}
/**
* Setup environment for every test case.
* @param hdfsConf hdfs conf.
* @throws Exception
*/
public void clusterSetUp(Configuration hdfsConf) throws Exception {
clusterSetUp(false, hdfsConf);
}
/**
* Setup cluster environment.
* @param isHAEnabled if true, enable simple HA.
* @throws IOException
*/
private void clusterSetUp(boolean isHAEnabled, Configuration newConf)
throws Exception {
conf = newConf;
conf.set(
DFSConfigKeys.DFS_STORAGE_POLICY_SATISFIER_RECHECK_TIMEOUT_MILLIS_KEY,
"3000");
conf.set(DFSConfigKeys.DFS_STORAGE_POLICY_SATISFIER_MODE_KEY,
StoragePolicySatisfierMode.EXTERNAL.toString());
// Reduced refresh cycle to update latest datanodes.
conf.setLong(DFSConfigKeys.DFS_SPS_DATANODE_CACHE_REFRESH_INTERVAL_MS,
1000);
conf.setInt(
DFSConfigKeys.DFS_STORAGE_POLICY_SATISFIER_MAX_RETRY_ATTEMPTS_KEY, 20);
final int dnNumber = storageTypes.length;
final short replication = 3;
MiniDFSCluster.Builder clusterBuilder = new MiniDFSCluster.Builder(conf)
.storageTypes(storageTypes).storagesPerDatanode(3)
.numDataNodes(dnNumber);
if (isHAEnabled) {
clusterBuilder.nnTopology(MiniDFSNNTopology.simpleHATopology());
}
cluster = clusterBuilder.build();
cluster.waitActive();
if (isHAEnabled) {
cluster.transitionToActive(0);
fs = HATestUtil.configureFailoverFs(cluster, conf);
} else {
fs = cluster.getFileSystem();
}
nnc = DFSTestUtil.getNameNodeConnector(conf,
HdfsServerConstants.MOVER_ID_PATH, 1, false);
sps = new StoragePolicySatisfier(conf);
ctxt = new ExternalSPSContext(sps, nnc);
sps.init(ctxt);
sps.start(StoragePolicySatisfierMode.EXTERNAL);
createTestFiles(fs, replication);
}
/**
* Setup test files for testing.
* @param dfs
* @param replication
* @throws Exception
*/
private void createTestFiles(DistributedFileSystem dfs,
short replication) throws Exception {
DFSTestUtil.createFile(dfs, testFile, 1024L, replication, 0L);
DFSTestUtil.createFile(dfs, parentFile, 1024L, replication, 0L);
DFSTestUtil.createFile(dfs, childFile, 1024L, replication, 0L);
DFSTestUtil.waitReplication(dfs, testFile, replication);
DFSTestUtil.waitReplication(dfs, parentFile, replication);
DFSTestUtil.waitReplication(dfs, childFile, replication);
}
/**
* Tear down environment for every test case.
* @throws IOException
*/
private void clusterShutdown() throws IOException{
if(fs != null) {
fs.close();
fs = null;
}
if(cluster != null) {
cluster.shutdown(true);
cluster = null;
}
if (sps != null) {
sps.stopGracefully();
}
}
/**
* While satisfying file/directory, trigger the cluster's checkpoint to
* make sure satisfier persistence work as expected. This test case runs
* as below:
* 1. use satisfyStoragePolicy and add xAttr to the file.
* 2. do the checkpoint by secondary NameNode.
* 3. restart the cluster immediately.
* 4. make sure all the storage policies are satisfied.
* @throws Exception
*/
@Test(timeout = 300000)
public void testWithCheckpoint() throws Exception {
SecondaryNameNode secondary = null;
try {
clusterSetUp();
fs.setStoragePolicy(testFile, WARM);
fs.satisfyStoragePolicy(testFile);
// Start the checkpoint.
conf.set(
DFSConfigKeys.DFS_NAMENODE_SECONDARY_HTTP_ADDRESS_KEY, "0.0.0.0:0");
secondary = new SecondaryNameNode(conf);
secondary.doCheckpoint();
restartCluster();
DFSTestUtil.waitExpectedStorageType(
testFileName, StorageType.DISK, 1, timeout, fs);
DFSTestUtil.waitExpectedStorageType(
testFileName, StorageType.ARCHIVE, 2, timeout, fs);
fs.setStoragePolicy(parentDir, COLD);
fs.satisfyStoragePolicy(parentDir);
DFSTestUtil.waitExpectedStorageType(
parentFileName, StorageType.ARCHIVE, 3, timeout, fs);
DFSTestUtil.waitExpectedStorageType(
childFileName, StorageType.ARCHIVE, 3, timeout, fs);
} finally {
if (secondary != null) {
secondary.shutdown();
}
clusterShutdown();
}
}
/**
* Tests to verify satisfier persistence working well with multiple
* restarts operations. This test case runs as below:
* 1. satisfy the storage policy of file1.
* 2. restart the cluster.
* 3. check whether all the blocks are satisfied.
* 4. satisfy the storage policy of file2.
* 5. restart the cluster.
* 6. check whether all the blocks are satisfied.
* @throws Exception
*/
@Test(timeout = 300000)
public void testWithRestarts() throws Exception {
try {
clusterSetUp();
fs.setStoragePolicy(testFile, ONE_SSD);
fs.satisfyStoragePolicy(testFile);
restartCluster();
DFSTestUtil.waitExpectedStorageType(
testFileName, StorageType.SSD, 1, timeout, fs);
DFSTestUtil.waitExpectedStorageType(
testFileName, StorageType.DISK, 2, timeout, fs);
// test directory
fs.setStoragePolicy(parentDir, COLD);
fs.satisfyStoragePolicy(parentDir);
restartCluster();
DFSTestUtil.waitExpectedStorageType(
parentFileName, StorageType.ARCHIVE, 3, timeout, fs);
DFSTestUtil.waitExpectedStorageType(
childFileName, StorageType.ARCHIVE, 3, timeout, fs);
} finally {
clusterShutdown();
}
}
/**
* Tests to verify SPS xattr will be removed if the satisfy work has
* been finished, expect that the method satisfyStoragePolicy can be
* invoked on the same file again after the block movement has been
* finished:
* 1. satisfy storage policy of file1.
* 2. wait until storage policy is satisfied.
* 3. satisfy storage policy of file1 again
* 4. make sure step 3 works as expected.
* @throws Exception
*/
@Test(timeout = 300000)
public void testMultipleSatisfyStoragePolicy() throws Exception {
try {
// Lower block movement check for testing.
conf = new HdfsConfiguration();
final long minCheckTimeout = 500; // minimum value
conf.setLong(
DFSConfigKeys.DFS_STORAGE_POLICY_SATISFIER_RECHECK_TIMEOUT_MILLIS_KEY,
minCheckTimeout);
clusterSetUp(conf);
fs.setStoragePolicy(testFile, ONE_SSD);
fs.satisfyStoragePolicy(testFile);
DFSTestUtil.waitExpectedStorageType(
testFileName, StorageType.SSD, 1, timeout, fs);
DFSTestUtil.waitExpectedStorageType(
testFileName, StorageType.DISK, 2, timeout, fs);
// Make sure satisfy xattr has been removed.
DFSTestUtil.waitForXattrRemoved(testFileName,
XATTR_SATISFY_STORAGE_POLICY, cluster.getNamesystem(), 30000);
fs.setStoragePolicy(testFile, COLD);
fs.satisfyStoragePolicy(testFile);
DFSTestUtil.waitExpectedStorageType(
testFileName, StorageType.ARCHIVE, 3, timeout, fs);
} finally {
clusterShutdown();
}
}
/**
* Tests to verify SPS xattr is removed after SPS is dropped,
* expect that if the SPS is disabled/dropped, the SPS
* xattr should be removed accordingly:
* 1. satisfy storage policy of file1.
* 2. drop SPS thread in block manager.
* 3. make sure sps xattr is removed.
* @throws Exception
*/
@Test(timeout = 300000000)
public void testDropSPS() throws Exception {
try {
clusterSetUp();
fs.setStoragePolicy(testFile, ONE_SSD);
fs.satisfyStoragePolicy(testFile);
cluster.getNamesystem().getBlockManager().getSPSManager()
.changeModeEvent(StoragePolicySatisfierMode.NONE);
// Make sure satisfy xattr has been removed.
DFSTestUtil.waitForXattrRemoved(testFileName,
XATTR_SATISFY_STORAGE_POLICY, cluster.getNamesystem(), 30000);
} finally {
clusterShutdown();
}
}
/**
* Tests that Xattrs should be cleaned if all blocks already satisfied.
*
* @throws Exception
*/
@Test(timeout = 300000)
public void testSPSShouldNotLeakXattrIfStorageAlreadySatisfied()
throws Exception {
try {
clusterSetUp();
DFSTestUtil.waitExpectedStorageType(testFileName, StorageType.DISK, 3,
timeout, fs);
fs.satisfyStoragePolicy(testFile);
DFSTestUtil.waitExpectedStorageType(testFileName, StorageType.DISK, 3,
timeout, fs);
// Make sure satisfy xattr has been removed.
DFSTestUtil.waitForXattrRemoved(testFileName,
XATTR_SATISFY_STORAGE_POLICY, cluster.getNamesystem(), 30000);
} finally {
clusterShutdown();
}
}
/**
* Test loading of SPS xAttrs from the edits log when satisfyStoragePolicy
* called on child file and parent directory.
* 1. Create one directory and create one child file.
* 2. Set storage policy for child file and call
* satisfyStoragePolicy.
* 3. wait for SPS to remove xAttr for file child file.
* 4. Set storage policy for parent directory and call
* satisfyStoragePolicy.
* 5. restart the namenode.
* NameNode should be started successfully.
*/
@Test(timeout = 300000)
public void testNameNodeRestartWhenSPSCalledOnChildFileAndParentDir()
throws Exception {
try {
clusterSetUp();
fs.setStoragePolicy(childFile, "COLD");
fs.satisfyStoragePolicy(childFile);
DFSTestUtil.waitExpectedStorageType(childFile.toUri().getPath(),
StorageType.ARCHIVE, 3, 30000, cluster.getFileSystem());
// wait for SPS to remove Xattr from file
Thread.sleep(30000);
fs.setStoragePolicy(childDir, "COLD");
fs.satisfyStoragePolicy(childDir);
try {
cluster.restartNameNodes();
} catch (Exception e) {
assertFalse(e.getMessage().contains(
"Cannot request to call satisfy storage policy"));
}
} finally {
clusterShutdown();
}
}
/**
* Test SPS when satisfyStoragePolicy called on child file and
* parent directory.
* 1. Create one parent directory and child directory.
* 2. Create some file in both the directory.
* 3. Set storage policy for parent directory and call
* satisfyStoragePolicy.
* 4. Set storage policy for child directory and call
* satisfyStoragePolicy.
* 5. restart the namenode.
* All the file blocks should satisfy the policy.
*/
@Test(timeout = 300000)
public void testSPSOnChildAndParentDirectory() throws Exception {
try {
clusterSetUp();
fs.setStoragePolicy(parentDir, "COLD");
fs.satisfyStoragePolicy(childDir);
DFSTestUtil.waitExpectedStorageType(childFileName, StorageType.ARCHIVE,
3, 30000, cluster.getFileSystem());
fs.satisfyStoragePolicy(parentDir);
DFSTestUtil.waitExpectedStorageType(parentFileName, StorageType.ARCHIVE,
3, 30000, cluster.getFileSystem());
} finally {
clusterShutdown();
}
}
/**
* Test SPS xAttr on directory. xAttr should be removed from the directory
* once all the files blocks moved to specific storage.
*/
@Test(timeout = 300000)
public void testSPSxAttrWhenSpsCalledForDir() throws Exception {
try {
clusterSetUp();
Path parent = new Path("/parent");
// create parent dir
fs.mkdirs(parent);
// create 10 child files
for (int i = 0; i < 5; i++) {
DFSTestUtil.createFile(fs, new Path(parent, "f" + i), 1024, (short) 3,
0);
}
// Set storage policy for parent directory
fs.setStoragePolicy(parent, "COLD");
// Stop one DN so we can check the SPS xAttr for directory.
DataNodeProperties stopDataNode = cluster.stopDataNode(0);
fs.satisfyStoragePolicy(parent);
// Check xAttr for parent directory
FSNamesystem namesystem = cluster.getNamesystem();
INode inode = namesystem.getFSDirectory().getINode("/parent");
XAttrFeature f = inode.getXAttrFeature();
assertTrue("SPS xAttr should be exist",
f.getXAttr(XATTR_SATISFY_STORAGE_POLICY) != null);
// check for the child, SPS xAttr should not be there
for (int i = 0; i < 5; i++) {
inode = namesystem.getFSDirectory().getINode("/parent/f" + i);
f = inode.getXAttrFeature();
assertTrue(f == null);
}
cluster.restartDataNode(stopDataNode, false);
// wait and check all the file block moved in ARCHIVE
for (int i = 0; i < 5; i++) {
DFSTestUtil.waitExpectedStorageType("/parent/f" + i,
StorageType.ARCHIVE, 3, 30000, cluster.getFileSystem());
}
DFSTestUtil.waitForXattrRemoved("/parent", XATTR_SATISFY_STORAGE_POLICY,
namesystem, 10000);
} finally {
clusterShutdown();
}
}
/**
* Test SPS xAttr on file. xAttr should be removed from the file
* once all the blocks moved to specific storage.
*/
@Test(timeout = 300000)
public void testSPSxAttrWhenSpsCalledForFile() throws Exception {
try {
clusterSetUp();
Path file = new Path("/file");
DFSTestUtil.createFile(fs, file, 1024, (short) 3, 0);
// Set storage policy for file
fs.setStoragePolicy(file, "COLD");
// Stop one DN so we can check the SPS xAttr for file.
DataNodeProperties stopDataNode = cluster.stopDataNode(0);
fs.satisfyStoragePolicy(file);
// Check xAttr for parent directory
FSNamesystem namesystem = cluster.getNamesystem();
INode inode = namesystem.getFSDirectory().getINode("/file");
XAttrFeature f = inode.getXAttrFeature();
assertTrue("SPS xAttr should be exist",
f.getXAttr(XATTR_SATISFY_STORAGE_POLICY) != null);
cluster.restartDataNode(stopDataNode, false);
// wait and check all the file block moved in ARCHIVE
DFSTestUtil.waitExpectedStorageType("/file", StorageType.ARCHIVE, 3,
30000, cluster.getFileSystem());
GenericTestUtils.waitFor(new Supplier<Boolean>() {
@Override
public Boolean get() {
List<XAttr> existingXAttrs = XAttrStorage.readINodeXAttrs(inode);
return !existingXAttrs.contains(XATTR_SATISFY_STORAGE_POLICY);
}
}, 100, 10000);
} finally {
clusterShutdown();
}
}
/**
* Restart the hole env and trigger the DataNode's heart beats.
* @throws Exception
*/
private void restartCluster() throws Exception {
cluster.restartDataNodes();
cluster.restartNameNodes();
cluster.waitActive();
cluster.triggerHeartbeats();
}
}