|
@@ -0,0 +1,375 @@
|
|
|
+/**
|
|
|
+ * 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;
|
|
|
+
|
|
|
+import static org.junit.Assert.assertEquals;
|
|
|
+import static org.junit.Assert.assertTrue;
|
|
|
+
|
|
|
+import java.io.IOException;
|
|
|
+import java.util.ArrayList;
|
|
|
+import java.util.HashMap;
|
|
|
+import java.util.List;
|
|
|
+import java.util.Map;
|
|
|
+import java.util.Random;
|
|
|
+
|
|
|
+import org.apache.commons.logging.Log;
|
|
|
+import org.apache.commons.logging.LogFactory;
|
|
|
+import org.apache.hadoop.conf.Configuration;
|
|
|
+import org.apache.hadoop.fs.CommonConfigurationKeys;
|
|
|
+import org.apache.hadoop.fs.FSDataOutputStream;
|
|
|
+import org.apache.hadoop.fs.FileSystem;
|
|
|
+import org.apache.hadoop.fs.Path;
|
|
|
+import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
|
|
|
+import org.apache.hadoop.hdfs.protocol.DatanodeInfo.AdminStates;
|
|
|
+import org.apache.hadoop.hdfs.protocol.HdfsConstants.DatanodeReportType;
|
|
|
+import org.apache.hadoop.hdfs.server.blockmanagement.CombinedHostFileManager;
|
|
|
+import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeDescriptor;
|
|
|
+import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeManager;
|
|
|
+import org.apache.hadoop.hdfs.server.blockmanagement.HostConfigManager;
|
|
|
+import org.apache.hadoop.hdfs.server.namenode.FSNamesystem;
|
|
|
+import org.apache.hadoop.hdfs.server.namenode.NameNodeAdapter;
|
|
|
+import org.apache.hadoop.hdfs.util.HostsFileWriter;
|
|
|
+import org.junit.After;
|
|
|
+import org.junit.Before;
|
|
|
+
|
|
|
+/**
|
|
|
+ * This class provide utilities for testing of the admin operations of nodes.
|
|
|
+ */
|
|
|
+public class AdminStatesBaseTest {
|
|
|
+ public static final Log LOG = LogFactory.getLog(AdminStatesBaseTest.class);
|
|
|
+ static final long seed = 0xDEADBEEFL;
|
|
|
+ static final int blockSize = 8192;
|
|
|
+ static final int fileSize = 16384;
|
|
|
+ static final int HEARTBEAT_INTERVAL = 1; // heartbeat interval in seconds
|
|
|
+ static final int BLOCKREPORT_INTERVAL_MSEC = 1000; //block report in msec
|
|
|
+ static final int NAMENODE_REPLICATION_INTERVAL = 1; //replication interval
|
|
|
+
|
|
|
+ final private Random myrand = new Random();
|
|
|
+
|
|
|
+ private HostsFileWriter hostsFileWriter;
|
|
|
+ private Configuration conf;
|
|
|
+ private MiniDFSCluster cluster = null;
|
|
|
+ private boolean useCombinedHostFileManager = false;
|
|
|
+
|
|
|
+ protected void setUseCombinedHostFileManager() {
|
|
|
+ useCombinedHostFileManager = true;
|
|
|
+ }
|
|
|
+
|
|
|
+ protected Configuration getConf() {
|
|
|
+ return conf;
|
|
|
+ }
|
|
|
+
|
|
|
+ protected MiniDFSCluster getCluster() {
|
|
|
+ return cluster;
|
|
|
+ }
|
|
|
+
|
|
|
+ @Before
|
|
|
+ public void setup() throws IOException {
|
|
|
+ // Set up the hosts/exclude files.
|
|
|
+ hostsFileWriter = new HostsFileWriter();
|
|
|
+ conf = new HdfsConfiguration();
|
|
|
+
|
|
|
+ if (useCombinedHostFileManager) {
|
|
|
+ conf.setClass(DFSConfigKeys.DFS_NAMENODE_HOSTS_PROVIDER_CLASSNAME_KEY,
|
|
|
+ CombinedHostFileManager.class, HostConfigManager.class);
|
|
|
+ }
|
|
|
+
|
|
|
+ // Setup conf
|
|
|
+ conf.setBoolean(DFSConfigKeys.DFS_NAMENODE_REPLICATION_CONSIDERLOAD_KEY,
|
|
|
+ false);
|
|
|
+ conf.setInt(DFSConfigKeys.DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY,
|
|
|
+ 200);
|
|
|
+ conf.setInt(DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_KEY, HEARTBEAT_INTERVAL);
|
|
|
+ conf.setInt(DFSConfigKeys.DFS_BLOCKREPORT_INTERVAL_MSEC_KEY,
|
|
|
+ BLOCKREPORT_INTERVAL_MSEC);
|
|
|
+ conf.setInt(DFSConfigKeys.DFS_NAMENODE_REPLICATION_INTERVAL_KEY,
|
|
|
+ NAMENODE_REPLICATION_INTERVAL);
|
|
|
+ conf.setInt(DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_INTERVAL_KEY, 1);
|
|
|
+
|
|
|
+ hostsFileWriter.initialize(conf, "temp/admin");
|
|
|
+ }
|
|
|
+
|
|
|
+ @After
|
|
|
+ public void teardown() throws IOException {
|
|
|
+ hostsFileWriter.cleanup();
|
|
|
+ shutdownCluster();
|
|
|
+ }
|
|
|
+
|
|
|
+ protected void writeFile(FileSystem fileSys, Path name, int repl)
|
|
|
+ throws IOException {
|
|
|
+ writeFile(fileSys, name, repl, 2);
|
|
|
+ }
|
|
|
+
|
|
|
+ protected void writeFile(FileSystem fileSys, Path name, int repl,
|
|
|
+ int numOfBlocks) throws IOException {
|
|
|
+ writeFile(fileSys, name, repl, numOfBlocks, true);
|
|
|
+ }
|
|
|
+
|
|
|
+ protected FSDataOutputStream writeFile(FileSystem fileSys, Path name,
|
|
|
+ int repl, int numOfBlocks, boolean completeFile)
|
|
|
+ throws IOException {
|
|
|
+ // create and write a file that contains two blocks of data
|
|
|
+ FSDataOutputStream stm = fileSys.create(name, true, fileSys.getConf()
|
|
|
+ .getInt(CommonConfigurationKeys.IO_FILE_BUFFER_SIZE_KEY, 4096),
|
|
|
+ (short) repl, blockSize);
|
|
|
+ byte[] buffer = new byte[blockSize*numOfBlocks];
|
|
|
+ Random rand = new Random(seed);
|
|
|
+ rand.nextBytes(buffer);
|
|
|
+ stm.write(buffer);
|
|
|
+ LOG.info("Created file " + name + " with " + repl + " replicas.");
|
|
|
+ if (completeFile) {
|
|
|
+ stm.close();
|
|
|
+ return null;
|
|
|
+ } else {
|
|
|
+ // Do not close stream, return it
|
|
|
+ // so that it is not garbage collected
|
|
|
+ return stm;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * decommission the DN or put the DN into maintenance for datanodeUuid or one
|
|
|
+ * random node if datanodeUuid is null.
|
|
|
+ * And wait for the node to reach the given {@code waitForState}.
|
|
|
+ */
|
|
|
+ protected DatanodeInfo takeNodeOutofService(int nnIndex,
|
|
|
+ String datanodeUuid, long maintenanceExpirationInMS,
|
|
|
+ ArrayList<DatanodeInfo> decommissionedNodes,
|
|
|
+ AdminStates waitForState) throws IOException {
|
|
|
+ return takeNodeOutofService(nnIndex, datanodeUuid,
|
|
|
+ maintenanceExpirationInMS, decommissionedNodes, null, waitForState);
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * decommission the DN or put the DN to maintenance set by datanodeUuid
|
|
|
+ * Pick randome node if datanodeUuid == null
|
|
|
+ * wait for the node to reach the given {@code waitForState}.
|
|
|
+ */
|
|
|
+ protected DatanodeInfo takeNodeOutofService(int nnIndex,
|
|
|
+ String datanodeUuid, long maintenanceExpirationInMS,
|
|
|
+ List<DatanodeInfo> decommissionedNodes,
|
|
|
+ Map<DatanodeInfo, Long> inMaintenanceNodes, AdminStates waitForState)
|
|
|
+ throws IOException {
|
|
|
+ DFSClient client = getDfsClient(nnIndex);
|
|
|
+ DatanodeInfo[] info = client.datanodeReport(DatanodeReportType.ALL);
|
|
|
+ boolean isDecommissionRequest =
|
|
|
+ waitForState == AdminStates.DECOMMISSION_INPROGRESS ||
|
|
|
+ waitForState == AdminStates.DECOMMISSIONED;
|
|
|
+
|
|
|
+ //
|
|
|
+ // pick one datanode randomly unless the caller specifies one.
|
|
|
+ //
|
|
|
+ int index = 0;
|
|
|
+ if (datanodeUuid == null) {
|
|
|
+ boolean found = false;
|
|
|
+ while (!found) {
|
|
|
+ index = myrand.nextInt(info.length);
|
|
|
+ if ((isDecommissionRequest && !info[index].isDecommissioned()) ||
|
|
|
+ (!isDecommissionRequest && !info[index].isInMaintenance())) {
|
|
|
+ found = true;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ } else {
|
|
|
+ // The caller specifies a DN
|
|
|
+ for (; index < info.length; index++) {
|
|
|
+ if (info[index].getDatanodeUuid().equals(datanodeUuid)) {
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if (index == info.length) {
|
|
|
+ throw new IOException("invalid datanodeUuid " + datanodeUuid);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ String nodename = info[index].getXferAddr();
|
|
|
+ LOG.info("Taking node: " + nodename + " out of service");
|
|
|
+
|
|
|
+ ArrayList<String> decommissionNodes = new ArrayList<String>();
|
|
|
+ if (decommissionedNodes != null) {
|
|
|
+ for (DatanodeInfo dn : decommissionedNodes) {
|
|
|
+ decommissionNodes.add(dn.getName());
|
|
|
+ }
|
|
|
+ }
|
|
|
+ Map<String, Long> maintenanceNodes = new HashMap<>();
|
|
|
+ if (inMaintenanceNodes != null) {
|
|
|
+ for (Map.Entry<DatanodeInfo, Long> dn :
|
|
|
+ inMaintenanceNodes.entrySet()) {
|
|
|
+ maintenanceNodes.put(dn.getKey().getName(), dn.getValue());
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ if (isDecommissionRequest) {
|
|
|
+ decommissionNodes.add(nodename);
|
|
|
+ } else {
|
|
|
+ maintenanceNodes.put(nodename, maintenanceExpirationInMS);
|
|
|
+ }
|
|
|
+
|
|
|
+ // write node names into the json host file.
|
|
|
+ hostsFileWriter.initOutOfServiceHosts(decommissionNodes, maintenanceNodes);
|
|
|
+ refreshNodes(nnIndex);
|
|
|
+ DatanodeInfo ret = NameNodeAdapter.getDatanode(
|
|
|
+ cluster.getNamesystem(nnIndex), info[index]);
|
|
|
+ waitNodeState(ret, waitForState);
|
|
|
+ return ret;
|
|
|
+ }
|
|
|
+
|
|
|
+ /* Ask a specific NN to put the datanode in service and wait for it
|
|
|
+ * to reach the NORMAL state.
|
|
|
+ */
|
|
|
+ protected void putNodeInService(int nnIndex,
|
|
|
+ DatanodeInfo outOfServiceNode) throws IOException {
|
|
|
+ LOG.info("Putting node: " + outOfServiceNode + " in service");
|
|
|
+ ArrayList<String> decommissionNodes = new ArrayList<>();
|
|
|
+ Map<String, Long> maintenanceNodes = new HashMap<>();
|
|
|
+
|
|
|
+ DatanodeManager dm =
|
|
|
+ cluster.getNamesystem(nnIndex).getBlockManager().getDatanodeManager();
|
|
|
+ List<DatanodeDescriptor> nodes =
|
|
|
+ dm.getDatanodeListForReport(DatanodeReportType.ALL);
|
|
|
+ for (DatanodeDescriptor node : nodes) {
|
|
|
+ if (node.isMaintenance()) {
|
|
|
+ maintenanceNodes.put(node.getName(),
|
|
|
+ node.getMaintenanceExpireTimeInMS());
|
|
|
+ } else if (node.isDecommissionInProgress() || node.isDecommissioned()) {
|
|
|
+ decommissionNodes.add(node.getName());
|
|
|
+ }
|
|
|
+ }
|
|
|
+ decommissionNodes.remove(outOfServiceNode.getName());
|
|
|
+ maintenanceNodes.remove(outOfServiceNode.getName());
|
|
|
+
|
|
|
+ hostsFileWriter.initOutOfServiceHosts(decommissionNodes, maintenanceNodes);
|
|
|
+ refreshNodes(nnIndex);
|
|
|
+ waitNodeState(outOfServiceNode, AdminStates.NORMAL);
|
|
|
+ }
|
|
|
+
|
|
|
+ protected void putNodeInService(int nnIndex,
|
|
|
+ String datanodeUuid) throws IOException {
|
|
|
+ DatanodeInfo datanodeInfo =
|
|
|
+ getDatanodeDesriptor(cluster.getNamesystem(nnIndex), datanodeUuid);
|
|
|
+ putNodeInService(nnIndex, datanodeInfo);
|
|
|
+ }
|
|
|
+
|
|
|
+ /*
|
|
|
+ * Wait till node is transitioned to the expected state.
|
|
|
+ */
|
|
|
+ protected void waitNodeState(DatanodeInfo node,
|
|
|
+ AdminStates state) {
|
|
|
+ boolean done = state == node.getAdminState();
|
|
|
+ while (!done) {
|
|
|
+ LOG.info("Waiting for node " + node + " to change state to "
|
|
|
+ + state + " current state: " + node.getAdminState());
|
|
|
+ try {
|
|
|
+ Thread.sleep(HEARTBEAT_INTERVAL * 500);
|
|
|
+ } catch (InterruptedException e) {
|
|
|
+ // nothing
|
|
|
+ }
|
|
|
+ done = state == node.getAdminState();
|
|
|
+ }
|
|
|
+ LOG.info("node " + node + " reached the state " + state);
|
|
|
+ }
|
|
|
+
|
|
|
+ protected void initIncludeHost(String hostNameAndPort) throws IOException {
|
|
|
+ hostsFileWriter.initIncludeHost(hostNameAndPort);
|
|
|
+ }
|
|
|
+
|
|
|
+ protected void initIncludeHosts(String[] hostNameAndPorts)
|
|
|
+ throws IOException {
|
|
|
+ hostsFileWriter.initIncludeHosts(hostNameAndPorts);
|
|
|
+ }
|
|
|
+
|
|
|
+ protected void initExcludeHost(String hostNameAndPort) throws IOException {
|
|
|
+ hostsFileWriter.initExcludeHost(hostNameAndPort);
|
|
|
+ }
|
|
|
+
|
|
|
+ protected void initExcludeHosts(List<String> hostNameAndPorts)
|
|
|
+ throws IOException {
|
|
|
+ hostsFileWriter.initExcludeHosts(hostNameAndPorts);
|
|
|
+ }
|
|
|
+
|
|
|
+ /* Get DFSClient to the namenode */
|
|
|
+ protected DFSClient getDfsClient(final int nnIndex) throws IOException {
|
|
|
+ return new DFSClient(cluster.getNameNode(nnIndex).getNameNodeAddress(),
|
|
|
+ conf);
|
|
|
+ }
|
|
|
+
|
|
|
+ /* Validate cluster has expected number of datanodes */
|
|
|
+ protected static void validateCluster(DFSClient client, int numDNs)
|
|
|
+ throws IOException {
|
|
|
+ DatanodeInfo[] info = client.datanodeReport(DatanodeReportType.LIVE);
|
|
|
+ assertEquals("Number of Datanodes ", numDNs, info.length);
|
|
|
+ }
|
|
|
+
|
|
|
+ /** Start a MiniDFSCluster.
|
|
|
+ * @throws IOException */
|
|
|
+ protected void startCluster(int numNameNodes, int numDatanodes,
|
|
|
+ boolean setupHostsFile, long[] nodesCapacity,
|
|
|
+ boolean checkDataNodeHostConfig) throws IOException {
|
|
|
+ MiniDFSCluster.Builder builder = new MiniDFSCluster.Builder(conf)
|
|
|
+ .nnTopology(MiniDFSNNTopology.simpleFederatedTopology(numNameNodes))
|
|
|
+ .numDataNodes(numDatanodes);
|
|
|
+ if (setupHostsFile) {
|
|
|
+ builder.setupHostsFile(setupHostsFile);
|
|
|
+ }
|
|
|
+ if (nodesCapacity != null) {
|
|
|
+ builder.simulatedCapacities(nodesCapacity);
|
|
|
+ }
|
|
|
+ if (checkDataNodeHostConfig) {
|
|
|
+ builder.checkDataNodeHostConfig(checkDataNodeHostConfig);
|
|
|
+ }
|
|
|
+ cluster = builder.build();
|
|
|
+ cluster.waitActive();
|
|
|
+ for (int i = 0; i < numNameNodes; i++) {
|
|
|
+ DFSClient client = getDfsClient(i);
|
|
|
+ validateCluster(client, numDatanodes);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ protected void startCluster(int numNameNodes, int numDatanodes)
|
|
|
+ throws IOException {
|
|
|
+ startCluster(numNameNodes, numDatanodes, false, null, false);
|
|
|
+ }
|
|
|
+
|
|
|
+ protected void startSimpleHACluster(int numDatanodes) throws IOException {
|
|
|
+ cluster = new MiniDFSCluster.Builder(conf)
|
|
|
+ .nnTopology(MiniDFSNNTopology.simpleHATopology()).numDataNodes(
|
|
|
+ numDatanodes).build();
|
|
|
+ cluster.transitionToActive(0);
|
|
|
+ cluster.waitActive();
|
|
|
+ }
|
|
|
+
|
|
|
+ protected void shutdownCluster() {
|
|
|
+ if (cluster != null) {
|
|
|
+ cluster.shutdown();
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ protected void refreshNodes(final int nnIndex) throws IOException {
|
|
|
+ cluster.getNamesystem(nnIndex).getBlockManager().getDatanodeManager().
|
|
|
+ refreshNodes(conf);
|
|
|
+ }
|
|
|
+
|
|
|
+ protected DatanodeDescriptor getDatanodeDesriptor(
|
|
|
+ final FSNamesystem ns, final String datanodeUuid) {
|
|
|
+ return ns.getBlockManager().getDatanodeManager().getDatanode(datanodeUuid);
|
|
|
+ }
|
|
|
+
|
|
|
+ protected void cleanupFile(FileSystem fileSys, Path name) throws IOException {
|
|
|
+ assertTrue(fileSys.exists(name));
|
|
|
+ fileSys.delete(name, true);
|
|
|
+ assertTrue(!fileSys.exists(name));
|
|
|
+ }
|
|
|
+}
|