TestBlockStatsMXBean.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
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* 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.blockmanagement;
import static org.apache.hadoop.test.PlatformAssumptions.assumeNotWindows;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.net.URL;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import org.apache.hadoop.fs.FSDataOutputStream;
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.DistributedFileSystem;
import org.apache.hadoop.hdfs.HdfsConfiguration;
import org.apache.hadoop.hdfs.MiniDFSCluster;
import org.apache.hadoop.hdfs.server.datanode.DataNodeTestUtils;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.test.GenericTestUtils;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.eclipse.jetty.util.ajax.JSON;
import org.junit.rules.Timeout;
/**
* Class for testing {@link BlockStatsMXBean} implementation
*/
public class TestBlockStatsMXBean {
private MiniDFSCluster cluster;
@Rule
public Timeout globalTimeout = new Timeout(300000);
@Before
public void setup() throws IOException {
HdfsConfiguration conf = new HdfsConfiguration();
conf.setTimeDuration(DFSConfigKeys.DFS_DATANODE_DISK_CHECK_MIN_GAP_KEY,
0, TimeUnit.MILLISECONDS);
cluster = null;
StorageType[][] types = new StorageType[7][];
for (int i=0; i<3; i++) {
types[i] = new StorageType[] {StorageType.RAM_DISK, StorageType.DISK};
}
for (int i=3; i< 5; i++) {
types[i] = new StorageType[] {StorageType.RAM_DISK, StorageType.ARCHIVE};
}
types[5] = new StorageType[] {StorageType.RAM_DISK, StorageType.ARCHIVE,
StorageType.ARCHIVE};
types[6] = new StorageType[]{StorageType.RAM_DISK, StorageType.NVDIMM};
cluster = new MiniDFSCluster.Builder(conf).numDataNodes(7).
storageTypes(types).storagesPerDatanode(3).build();
cluster.waitActive();
}
@After
public void tearDown() {
if (cluster != null) {
cluster.shutdown();
cluster = null;
}
}
@Test
public void testStorageTypeStats() throws Exception {
Map<StorageType, StorageTypeStats> storageTypeStatsMap =
cluster.getNamesystem().getBlockManager().getStorageTypeStats();
assertTrue(storageTypeStatsMap.containsKey(StorageType.RAM_DISK));
assertTrue(storageTypeStatsMap.containsKey(StorageType.DISK));
assertTrue(storageTypeStatsMap.containsKey(StorageType.ARCHIVE));
assertTrue(storageTypeStatsMap.containsKey(StorageType.NVDIMM));
StorageTypeStats storageTypeStats =
storageTypeStatsMap.get(StorageType.RAM_DISK);
assertEquals(7, storageTypeStats.getNodesInService());
storageTypeStats = storageTypeStatsMap.get(StorageType.DISK);
assertEquals(3, storageTypeStats.getNodesInService());
storageTypeStats = storageTypeStatsMap.get(StorageType.ARCHIVE);
assertEquals(3, storageTypeStats.getNodesInService());
storageTypeStats = storageTypeStatsMap.get(StorageType.NVDIMM);
assertEquals(1, storageTypeStats.getNodesInService());
}
protected static String readOutput(URL url) throws IOException {
StringBuilder out = new StringBuilder();
InputStream in = url.openConnection().getInputStream();
byte[] buffer = new byte[64 * 1024];
int len = in.read(buffer);
while (len > 0) {
out.append(new String(buffer, 0, len));
len = in.read(buffer);
}
return out.toString();
}
@Test
@SuppressWarnings("unchecked")
public void testStorageTypeStatsJMX() throws Exception {
URL baseUrl = new URL (cluster.getHttpUri(0));
String result = readOutput(new URL(baseUrl, "/jmx"));
Map<String, Object> stat = (Map<String, Object>) JSON.parse(result);
Object[] beans =(Object[]) stat.get("beans");
Map<String, Object> blockStats = null;
for (Object bean : beans) {
Map<String, Object> map = (Map<String, Object>) bean;
if (map.get("name").equals("Hadoop:service=NameNode,name=BlockStats")) {
blockStats = map;
}
}
assertNotNull(blockStats);
Object[] storageTypeStatsList =
(Object[])blockStats.get("StorageTypeStats");
assertNotNull(storageTypeStatsList);
assertEquals(4, storageTypeStatsList.length);
Set<String> typesPresent = new HashSet<> ();
for (Object obj : storageTypeStatsList) {
Map<String, Object> entry = (Map<String, Object>)obj;
String storageType = (String)entry.get("key");
Map<String,Object> storageTypeStats = (Map<String,Object>)entry.get("value");
typesPresent.add(storageType);
switch (storageType) {
case "ARCHIVE":
case "DISK":
assertEquals(3L, storageTypeStats.get("nodesInService"));
break;
case "RAM_DISK":
assertEquals(7L, storageTypeStats.get("nodesInService"));
break;
case "NVDIMM":
assertEquals(1L, storageTypeStats.get("nodesInService"));
break;
default:
fail();
}
}
assertTrue(typesPresent.contains("ARCHIVE"));
assertTrue(typesPresent.contains("DISK"));
assertTrue(typesPresent.contains("RAM_DISK"));
assertTrue(typesPresent.contains("NVDIMM"));
}
@Test
public void testStorageTypeStatsWhenStorageFailed() throws Exception {
// The test uses DataNodeTestUtils#injectDataDirFailure() to simulate
// volume failures which is currently not supported on Windows.
assumeNotWindows();
DFSTestUtil.createFile(cluster.getFileSystem(),
new Path("/blockStatsFile1"), 1024, (short) 1, 0L);
Map<StorageType, StorageTypeStats> storageTypeStatsMap = cluster
.getNamesystem().getBlockManager().getStorageTypeStats();
StorageTypeStats storageTypeStats = storageTypeStatsMap
.get(StorageType.RAM_DISK);
assertEquals(7, storageTypeStats.getNodesInService());
storageTypeStats = storageTypeStatsMap.get(StorageType.DISK);
assertEquals(3, storageTypeStats.getNodesInService());
storageTypeStats = storageTypeStatsMap.get(StorageType.ARCHIVE);
assertEquals(3, storageTypeStats.getNodesInService());
storageTypeStats = storageTypeStatsMap.get(StorageType.NVDIMM);
assertEquals(1, storageTypeStats.getNodesInService());
File dn1ArcVol1 = cluster.getInstanceStorageDir(0, 1);
File dn2ArcVol1 = cluster.getInstanceStorageDir(1, 1);
File dn3ArcVol1 = cluster.getInstanceStorageDir(2, 1);
File dn4ArcVol1 = cluster.getInstanceStorageDir(3, 1);
DataNodeTestUtils.injectDataDirFailure(dn1ArcVol1);
DataNodeTestUtils.injectDataDirFailure(dn2ArcVol1);
DataNodeTestUtils.injectDataDirFailure(dn3ArcVol1);
DataNodeTestUtils.injectDataDirFailure(dn4ArcVol1);
try {
DFSTestUtil.createFile(cluster.getFileSystem(), new Path(
"/blockStatsFile2"), 1024, (short) 1, 0L);
fail("Should throw exception, becuase no DISK storage available");
} catch (Exception e) {
assertTrue(e.getMessage().contains(
"could only be written to 0 of the 1 minReplication"));
}
// wait for heartbeat
Thread.sleep(6000);
storageTypeStatsMap = cluster.getNamesystem().getBlockManager()
.getStorageTypeStats();
assertFalse("StorageTypeStatsMap should not contain DISK Storage type",
storageTypeStatsMap.containsKey(StorageType.DISK));
DataNodeTestUtils.restoreDataDirFromFailure(dn1ArcVol1);
DataNodeTestUtils.restoreDataDirFromFailure(dn2ArcVol1);
DataNodeTestUtils.restoreDataDirFromFailure(dn3ArcVol1);
DataNodeTestUtils.restoreDataDirFromFailure(dn4ArcVol1);
for (int i = 0; i < 4; i++) {
cluster.restartDataNode(0, true);
}
// wait for heartbeat
Thread.sleep(6000);
storageTypeStatsMap = cluster.getNamesystem().getBlockManager()
.getStorageTypeStats();
storageTypeStats = storageTypeStatsMap.get(StorageType.RAM_DISK);
assertEquals(7, storageTypeStats.getNodesInService());
storageTypeStats = storageTypeStatsMap.get(StorageType.DISK);
assertEquals(3, storageTypeStats.getNodesInService());
storageTypeStats = storageTypeStatsMap.get(StorageType.ARCHIVE);
assertEquals(3, storageTypeStats.getNodesInService());
storageTypeStats = storageTypeStatsMap.get(StorageType.NVDIMM);
assertEquals(1, storageTypeStats.getNodesInService());
}
@Test
public void testStorageTypeLoad() throws Exception {
HeartbeatManager heartbeatManager =
cluster.getNamesystem().getBlockManager().getDatanodeManager()
.getHeartbeatManager();
Map<StorageType, StorageTypeStats> storageTypeStatsMap =
heartbeatManager.getStorageTypeStats();
DistributedFileSystem dfs = cluster.getFileSystem();
// Create a file with HOT storage policy.
Path hotSpDir = new Path("/HOT");
dfs.mkdir(hotSpDir, FsPermission.getDirDefault());
dfs.setStoragePolicy(hotSpDir, "HOT");
FSDataOutputStream hotSpFileStream =
dfs.create(new Path(hotSpDir, "hotFile"));
hotSpFileStream.write("Storage Policy Hot".getBytes());
hotSpFileStream.hflush();
// Create a file with COLD storage policy.
Path coldSpDir = new Path("/COLD");
dfs.mkdir(coldSpDir, FsPermission.getDirDefault());
dfs.setStoragePolicy(coldSpDir, "COLD");
FSDataOutputStream coldSpFileStream =
dfs.create(new Path(coldSpDir, "coldFile"));
coldSpFileStream.write("Writing to ARCHIVE storage type".getBytes());
coldSpFileStream.hflush();
// Trigger heartbeats manually to speed up the test.
cluster.triggerHeartbeats();
// The load would be 2*replication since both the
// write xceiver & packet responder threads are counted.
GenericTestUtils.waitFor(() -> storageTypeStatsMap.get(StorageType.DISK)
.getNodesInServiceXceiverCount() == 6, 100, 5000);
// The count for ARCHIVE should be independent of the value of DISK.
GenericTestUtils.waitFor(() -> storageTypeStatsMap.get(StorageType.ARCHIVE)
.getNodesInServiceXceiverCount() == 6, 100, 5000);
// The total count should stay unaffected, that is sum of load from all
// datanodes.
GenericTestUtils
.waitFor(() -> heartbeatManager.getInServiceXceiverCount() == 12, 100,
5000);
IOUtils.closeStreams(hotSpFileStream, coldSpFileStream);
}
@Test
public void testStorageTypePercentJMX() throws Exception {
URL baseUrl = new URL(cluster.getHttpUri(0));
String result = readOutput(new URL(baseUrl, "/jmx"));
Map<String, Object> stat = (Map<String, Object>) JSON.parse(result);
Object[] beans = (Object[]) stat.get("beans");
Map<String, Object> blockStats = null;
for (Object bean : beans) {
Map<String, Object> map = (Map<String, Object>) bean;
if (map.get("name").equals("Hadoop:service=NameNode,name=BlockStats")) {
blockStats = map;
}
}
assertNotNull(blockStats);
Object[] storageTypeStatsList =
(Object[]) blockStats.get("StorageTypeStats");
assertNotNull(storageTypeStatsList);
Map<String, Object> entry = (Map<String, Object>) storageTypeStatsList[0];
Map<String, Object> storageTypeStats = (Map<String, Object>) entry.get("value");
assertTrue(storageTypeStats.containsKey("percentUsed"));
assertTrue(storageTypeStats.containsKey("percentBlockPoolUsed"));
assertTrue(storageTypeStats.containsKey("percentRemaining"));
}
}