1
0

MiniDFSCluster.java 30 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779
  1. /**
  2. * Licensed to the Apache Software Foundation (ASF) under one
  3. * or more contributor license agreements. See the NOTICE file
  4. * distributed with this work for additional information
  5. * regarding copyright ownership. The ASF licenses this file
  6. * to you under the Apache License, Version 2.0 (the
  7. * "License"); you may not use this file except in compliance
  8. * with the License. You may obtain a copy of the License at
  9. *
  10. * http://www.apache.org/licenses/LICENSE-2.0
  11. *
  12. * Unless required by applicable law or agreed to in writing, software
  13. * distributed under the License is distributed on an "AS IS" BASIS,
  14. * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15. * See the License for the specific language governing permissions and
  16. * limitations under the License.
  17. */
  18. package org.apache.hadoop.dfs;
  19. import java.io.File;
  20. import java.io.IOException;
  21. import java.net.InetSocketAddress;
  22. import java.util.ArrayList;
  23. import java.util.Collection;
  24. import javax.security.auth.login.LoginException;
  25. import org.apache.hadoop.conf.Configuration;
  26. import org.apache.hadoop.net.*;
  27. import org.apache.hadoop.dfs.FSConstants.DatanodeReportType;
  28. import org.apache.hadoop.dfs.FSConstants.StartupOption;
  29. import org.apache.hadoop.fs.FileSystem;
  30. import org.apache.hadoop.fs.FileUtil;
  31. import org.apache.hadoop.security.*;
  32. import org.apache.hadoop.util.ToolRunner;
  33. /**
  34. * This class creates a single-process DFS cluster for junit testing.
  35. * The data directories for non-simulated DFS are under the testing directory.
  36. * For simulated data nodes, no underlying fs storage is used.
  37. */
  38. public class MiniDFSCluster {
  39. private class DataNodeProperties {
  40. DataNode datanode;
  41. Configuration conf;
  42. String[] dnArgs;
  43. DataNodeProperties(DataNode node, Configuration conf, String[] args) {
  44. this.datanode = node;
  45. this.conf = conf;
  46. this.dnArgs = args;
  47. }
  48. }
  49. private Configuration conf;
  50. private NameNode nameNode;
  51. private int numDataNodes;
  52. private ArrayList<DataNodeProperties> dataNodes =
  53. new ArrayList<DataNodeProperties>();
  54. private File base_dir;
  55. private File data_dir;
  56. private DNSToSwitchMapping dnsToSwitchMapping;
  57. /**
  58. * This null constructor is used only when wishing to start a data node cluster
  59. * without a name node (ie when the name node is started elsewhere).
  60. */
  61. public MiniDFSCluster() {
  62. }
  63. /**
  64. * Modify the config and start up the servers with the given operation.
  65. * Servers will be started on free ports.
  66. * <p>
  67. * The caller must manage the creation of NameNode and DataNode directories
  68. * and have already set dfs.name.dir and dfs.data.dir in the given conf.
  69. *
  70. * @param conf the base configuration to use in starting the servers. This
  71. * will be modified as necessary.
  72. * @param numDataNodes Number of DataNodes to start; may be zero
  73. * @param nameNodeOperation the operation with which to start the servers. If null
  74. * or StartupOption.FORMAT, then StartupOption.REGULAR will be used.
  75. */
  76. public MiniDFSCluster(Configuration conf,
  77. int numDataNodes,
  78. StartupOption nameNodeOperation) throws IOException {
  79. this(0, conf, numDataNodes, false, false, nameNodeOperation, null, null, null);
  80. }
  81. /**
  82. * Modify the config and start up the servers. The rpc and info ports for
  83. * servers are guaranteed to use free ports.
  84. * <p>
  85. * NameNode and DataNode directory creation and configuration will be
  86. * managed by this class.
  87. *
  88. * @param conf the base configuration to use in starting the servers. This
  89. * will be modified as necessary.
  90. * @param numDataNodes Number of DataNodes to start; may be zero
  91. * @param format if true, format the NameNode and DataNodes before starting up
  92. * @param racks array of strings indicating the rack that each DataNode is on
  93. */
  94. public MiniDFSCluster(Configuration conf,
  95. int numDataNodes,
  96. boolean format,
  97. String[] racks) throws IOException {
  98. this(0, conf, numDataNodes, format, true, null, racks, null, null);
  99. }
  100. /**
  101. * Modify the config and start up the servers. The rpc and info ports for
  102. * servers are guaranteed to use free ports.
  103. * <p>
  104. * NameNode and DataNode directory creation and configuration will be
  105. * managed by this class.
  106. *
  107. * @param conf the base configuration to use in starting the servers. This
  108. * will be modified as necessary.
  109. * @param numDataNodes Number of DataNodes to start; may be zero
  110. * @param format if true, format the NameNode and DataNodes before starting up
  111. * @param racks array of strings indicating the rack that each DataNode is on
  112. * @param hosts array of strings indicating the hostname for each DataNode
  113. */
  114. public MiniDFSCluster(Configuration conf,
  115. int numDataNodes,
  116. boolean format,
  117. String[] racks, String[] hosts) throws IOException {
  118. this(0, conf, numDataNodes, format, true, null, racks, hosts, null);
  119. }
  120. /**
  121. * NOTE: if possible, the other constructors that don't have nameNode port
  122. * parameter should be used as they will ensure that the servers use free ports.
  123. * <p>
  124. * Modify the config and start up the servers.
  125. *
  126. * @param nameNodePort suggestion for which rpc port to use. caller should
  127. * use getNameNodePort() to get the actual port used.
  128. * @param conf the base configuration to use in starting the servers. This
  129. * will be modified as necessary.
  130. * @param numDataNodes Number of DataNodes to start; may be zero
  131. * @param format if true, format the NameNode and DataNodes before starting up
  132. * @param manageDfsDirs if true, the data directories for servers will be
  133. * created and dfs.name.dir and dfs.data.dir will be set in the conf
  134. * @param operation the operation with which to start the servers. If null
  135. * or StartupOption.FORMAT, then StartupOption.REGULAR will be used.
  136. * @param racks array of strings indicating the rack that each DataNode is on
  137. */
  138. public MiniDFSCluster(int nameNodePort,
  139. Configuration conf,
  140. int numDataNodes,
  141. boolean format,
  142. boolean manageDfsDirs,
  143. StartupOption operation,
  144. String[] racks) throws IOException {
  145. this(nameNodePort, conf, numDataNodes, format, manageDfsDirs, operation,
  146. racks, null, null);
  147. }
  148. /**
  149. * NOTE: if possible, the other constructors that don't have nameNode port
  150. * parameter should be used as they will ensure that the servers use free ports.
  151. * <p>
  152. * Modify the config and start up the servers.
  153. *
  154. * @param nameNodePort suggestion for which rpc port to use. caller should
  155. * use getNameNodePort() to get the actual port used.
  156. * @param conf the base configuration to use in starting the servers. This
  157. * will be modified as necessary.
  158. * @param numDataNodes Number of DataNodes to start; may be zero
  159. * @param format if true, format the NameNode and DataNodes before starting up
  160. * @param manageDfsDirs if true, the data directories for servers will be
  161. * created and dfs.name.dir and dfs.data.dir will be set in the conf
  162. * @param operation the operation with which to start the servers. If null
  163. * or StartupOption.FORMAT, then StartupOption.REGULAR will be used.
  164. * @param racks array of strings indicating the rack that each DataNode is on
  165. * @param simulatedCapacities array of capacities of the simulated data nodes
  166. */
  167. public MiniDFSCluster(int nameNodePort,
  168. Configuration conf,
  169. int numDataNodes,
  170. boolean format,
  171. boolean manageDfsDirs,
  172. StartupOption operation,
  173. String[] racks,
  174. long[] simulatedCapacities) throws IOException {
  175. this(nameNodePort, conf, numDataNodes, format, manageDfsDirs, operation, racks, null,
  176. simulatedCapacities);
  177. }
  178. /**
  179. * NOTE: if possible, the other constructors that don't have nameNode port
  180. * parameter should be used as they will ensure that the servers use free ports.
  181. * <p>
  182. * Modify the config and start up the servers.
  183. *
  184. * @param nameNodePort suggestion for which rpc port to use. caller should
  185. * use getNameNodePort() to get the actual port used.
  186. * @param conf the base configuration to use in starting the servers. This
  187. * will be modified as necessary.
  188. * @param numDataNodes Number of DataNodes to start; may be zero
  189. * @param format if true, format the NameNode and DataNodes before starting up
  190. * @param manageDfsDirs if true, the data directories for servers will be
  191. * created and dfs.name.dir and dfs.data.dir will be set in the conf
  192. * @param operation the operation with which to start the servers. If null
  193. * or StartupOption.FORMAT, then StartupOption.REGULAR will be used.
  194. * @param racks array of strings indicating the rack that each DataNode is on
  195. * @param hosts array of strings indicating the hostnames of each DataNode
  196. * @param simulatedCapacities array of capacities of the simulated data nodes
  197. */
  198. public MiniDFSCluster(int nameNodePort,
  199. Configuration conf,
  200. int numDataNodes,
  201. boolean format,
  202. boolean manageDfsDirs,
  203. StartupOption operation,
  204. String[] racks, String hosts[],
  205. long[] simulatedCapacities) throws IOException {
  206. this.conf = conf;
  207. try {
  208. UserGroupInformation.setCurrentUGI(UnixUserGroupInformation.login(conf));
  209. } catch (LoginException e) {
  210. IOException ioe = new IOException();
  211. ioe.initCause(e);
  212. throw ioe;
  213. }
  214. base_dir = new File(System.getProperty("test.build.data"), "dfs/");
  215. data_dir = new File(base_dir, "data");
  216. // Setup the NameNode configuration
  217. FileSystem.setDefaultUri(conf, "hdfs://localhost:"+ Integer.toString(nameNodePort));
  218. conf.set("dfs.http.address", "0.0.0.0:0");
  219. if (manageDfsDirs) {
  220. conf.set("dfs.name.dir", new File(base_dir, "name1").getPath()+","+
  221. new File(base_dir, "name2").getPath());
  222. }
  223. int replication = conf.getInt("dfs.replication", 3);
  224. conf.setInt("dfs.replication", Math.min(replication, numDataNodes));
  225. conf.setInt("dfs.safemode.extension", 0);
  226. conf.setInt("dfs.namenode.decommission.interval", 3); // 3 second
  227. // Format and clean out DataNode directories
  228. if (format) {
  229. if (data_dir.exists() && !FileUtil.fullyDelete(data_dir)) {
  230. throw new IOException("Cannot remove data directory: " + data_dir);
  231. }
  232. NameNode.format(conf);
  233. }
  234. // Start the NameNode
  235. String[] args = (operation == null ||
  236. operation == StartupOption.FORMAT ||
  237. operation == StartupOption.REGULAR) ?
  238. new String[] {} : new String[] {"-"+operation.toString()};
  239. conf.setClass("topology.node.switch.mapping.impl",
  240. StaticMapping.class, DNSToSwitchMapping.class);
  241. nameNode = NameNode.createNameNode(args, conf);
  242. // Start the DataNodes
  243. startDataNodes(conf, numDataNodes, manageDfsDirs, operation, racks, hosts, simulatedCapacities);
  244. waitClusterUp();
  245. }
  246. /**
  247. * wait for the cluster to get out of
  248. * safemode.
  249. */
  250. public void waitClusterUp() {
  251. if (numDataNodes > 0) {
  252. while (!isClusterUp()) {
  253. try {
  254. System.err.println("Waiting for the Mini HDFS Cluster to start...");
  255. Thread.sleep(1000);
  256. } catch (InterruptedException e) {
  257. }
  258. }
  259. }
  260. }
  261. /**
  262. * Modify the config and start up additional DataNodes. The info port for
  263. * DataNodes is guaranteed to use a free port.
  264. *
  265. * Data nodes can run with the name node in the mini cluster or
  266. * a real name node. For example, running with a real name node is useful
  267. * when running simulated data nodes with a real name node.
  268. * If minicluster's name node is null assume that the conf has been
  269. * set with the right address:port of the name node.
  270. *
  271. * @param conf the base configuration to use in starting the DataNodes. This
  272. * will be modified as necessary.
  273. * @param numDataNodes Number of DataNodes to start; may be zero
  274. * @param manageDfsDirs if true, the data directories for DataNodes will be
  275. * created and dfs.data.dir will be set in the conf
  276. * @param operation the operation with which to start the DataNodes. If null
  277. * or StartupOption.FORMAT, then StartupOption.REGULAR will be used.
  278. * @param racks array of strings indicating the rack that each DataNode is on
  279. * @param hosts array of strings indicating the hostnames for each DataNode
  280. * @param simulatedCapacities array of capacities of the simulated data nodes
  281. *
  282. * @throws IllegalStateException if NameNode has been shutdown
  283. */
  284. public synchronized void startDataNodes(Configuration conf, int numDataNodes,
  285. boolean manageDfsDirs, StartupOption operation,
  286. String[] racks, String[] hosts,
  287. long[] simulatedCapacities) throws IOException {
  288. int curDatanodesNum = dataNodes.size();
  289. // for mincluster's the default initialDelay for BRs is 0
  290. if (conf.get("dfs.blockreport.initialDelay") == null) {
  291. conf.setLong("dfs.blockreport.initialDelay", 0);
  292. }
  293. // If minicluster's name node is null assume that the conf has been
  294. // set with the right address:port of the name node.
  295. //
  296. if (nameNode != null) { // set conf from the name node
  297. InetSocketAddress nnAddr = nameNode.getNameNodeAddress();
  298. int nameNodePort = nnAddr.getPort();
  299. FileSystem.setDefaultUri(conf,
  300. "hdfs://"+ nnAddr.getHostName() +
  301. ":" + Integer.toString(nameNodePort));
  302. }
  303. if (racks != null && numDataNodes > racks.length ) {
  304. throw new IllegalArgumentException( "The length of racks [" + racks.length
  305. + "] is less than the number of datanodes [" + numDataNodes + "].");
  306. }
  307. if (hosts != null && numDataNodes > hosts.length ) {
  308. throw new IllegalArgumentException( "The length of hosts [" + hosts.length
  309. + "] is less than the number of datanodes [" + numDataNodes + "].");
  310. }
  311. //Generate some hostnames if required
  312. if (racks != null && hosts == null) {
  313. System.out.println("Generating host names for datanodes");
  314. hosts = new String[numDataNodes];
  315. for (int i = curDatanodesNum; i < curDatanodesNum + numDataNodes; i++) {
  316. hosts[i - curDatanodesNum] = "host" + i + ".foo.com";
  317. }
  318. }
  319. if (simulatedCapacities != null
  320. && numDataNodes > simulatedCapacities.length) {
  321. throw new IllegalArgumentException( "The length of simulatedCapacities ["
  322. + simulatedCapacities.length
  323. + "] is less than the number of datanodes [" + numDataNodes + "].");
  324. }
  325. // Set up the right ports for the datanodes
  326. conf.set("dfs.datanode.address", "127.0.0.1:0");
  327. conf.set("dfs.datanode.http.address", "127.0.0.1:0");
  328. conf.set("dfs.datanode.ipc.address", "0.0.0.0:0");
  329. String[] args = (operation == null ||
  330. operation == StartupOption.FORMAT ||
  331. operation == StartupOption.REGULAR) ?
  332. null : new String[] {"-"+operation.toString()};
  333. String [] dnArgs = (operation == StartupOption.UPGRADE) ? null : args;
  334. for (int i = curDatanodesNum; i < curDatanodesNum+numDataNodes; i++) {
  335. Configuration dnConf = new Configuration(conf);
  336. if (manageDfsDirs) {
  337. File dir1 = new File(data_dir, "data"+(2*i+1));
  338. File dir2 = new File(data_dir, "data"+(2*i+2));
  339. dir1.mkdirs();
  340. dir2.mkdirs();
  341. if (!dir1.isDirectory() || !dir2.isDirectory()) {
  342. throw new IOException("Mkdirs failed to create directory for DataNode "
  343. + i + ": " + dir1 + " or " + dir2);
  344. }
  345. dnConf.set("dfs.data.dir", dir1.getPath() + "," + dir2.getPath());
  346. }
  347. if (simulatedCapacities != null) {
  348. dnConf.setBoolean("dfs.datanode.simulateddatastorage", true);
  349. dnConf.setLong(SimulatedFSDataset.CONFIG_PROPERTY_CAPACITY,
  350. simulatedCapacities[i-curDatanodesNum]);
  351. }
  352. System.out.println("Starting DataNode " + i + " with dfs.data.dir: "
  353. + dnConf.get("dfs.data.dir"));
  354. if (hosts != null) {
  355. dnConf.set("slave.host.name", hosts[i - curDatanodesNum]);
  356. System.out.println("Starting DataNode " + i + " with hostname set to: "
  357. + dnConf.get("slave.host.name"));
  358. }
  359. if (racks != null) {
  360. String name = hosts[i - curDatanodesNum];
  361. System.out.println("Adding node with hostname : " + name + " to rack "+
  362. racks[i-curDatanodesNum]);
  363. StaticMapping.addNodeToRack(name, racks[i-curDatanodesNum]);
  364. }
  365. Configuration newconf = new Configuration(dnConf); // save config
  366. if (hosts != null) {
  367. NetUtils.addStaticResolution(hosts[i - curDatanodesNum], "localhost");
  368. }
  369. DataNode dn = DataNode.instantiateDataNode(dnArgs, dnConf);
  370. //since the HDFS does things based on IP:port, we need to add the mapping
  371. //for IP:port to rackId
  372. String ipAddr = dn.getSelfAddr().getAddress().getHostAddress();
  373. if (racks != null) {
  374. int port = dn.getSelfAddr().getPort();
  375. System.out.println("Adding node with IP:port : " + ipAddr + ":" + port+
  376. " to rack " + racks[i-curDatanodesNum]);
  377. StaticMapping.addNodeToRack(ipAddr + ":" + port,
  378. racks[i-curDatanodesNum]);
  379. }
  380. DataNode.runDatanodeDaemon(dn);
  381. dataNodes.add(new DataNodeProperties(dn, newconf, dnArgs));
  382. }
  383. curDatanodesNum += numDataNodes;
  384. this.numDataNodes += numDataNodes;
  385. waitActive();
  386. }
  387. /**
  388. * Modify the config and start up the DataNodes. The info port for
  389. * DataNodes is guaranteed to use a free port.
  390. *
  391. * @param conf the base configuration to use in starting the DataNodes. This
  392. * will be modified as necessary.
  393. * @param numDataNodes Number of DataNodes to start; may be zero
  394. * @param manageDfsDirs if true, the data directories for DataNodes will be
  395. * created and dfs.data.dir will be set in the conf
  396. * @param operation the operation with which to start the DataNodes. If null
  397. * or StartupOption.FORMAT, then StartupOption.REGULAR will be used.
  398. * @param racks array of strings indicating the rack that each DataNode is on
  399. *
  400. * @throws IllegalStateException if NameNode has been shutdown
  401. */
  402. public void startDataNodes(Configuration conf, int numDataNodes,
  403. boolean manageDfsDirs, StartupOption operation,
  404. String[] racks
  405. ) throws IOException {
  406. startDataNodes( conf, numDataNodes, manageDfsDirs, operation, racks, null, null);
  407. }
  408. /**
  409. * Modify the config and start up additional DataNodes. The info port for
  410. * DataNodes is guaranteed to use a free port.
  411. *
  412. * Data nodes can run with the name node in the mini cluster or
  413. * a real name node. For example, running with a real name node is useful
  414. * when running simulated data nodes with a real name node.
  415. * If minicluster's name node is null assume that the conf has been
  416. * set with the right address:port of the name node.
  417. *
  418. * @param conf the base configuration to use in starting the DataNodes. This
  419. * will be modified as necessary.
  420. * @param numDataNodes Number of DataNodes to start; may be zero
  421. * @param manageDfsDirs if true, the data directories for DataNodes will be
  422. * created and dfs.data.dir will be set in the conf
  423. * @param operation the operation with which to start the DataNodes. If null
  424. * or StartupOption.FORMAT, then StartupOption.REGULAR will be used.
  425. * @param racks array of strings indicating the rack that each DataNode is on
  426. * @param simulatedCapacities array of capacities of the simulated data nodes
  427. *
  428. * @throws IllegalStateException if NameNode has been shutdown
  429. */
  430. public void startDataNodes(Configuration conf, int numDataNodes,
  431. boolean manageDfsDirs, StartupOption operation,
  432. String[] racks,
  433. long[] simulatedCapacities) throws IOException {
  434. startDataNodes(conf, numDataNodes, manageDfsDirs, operation, racks, null,
  435. simulatedCapacities);
  436. }
  437. /**
  438. * If the NameNode is running, attempt to finalize a previous upgrade.
  439. * When this method return, the NameNode should be finalized, but
  440. * DataNodes may not be since that occurs asynchronously.
  441. *
  442. * @throws IllegalStateException if the Namenode is not running.
  443. */
  444. public void finalizeCluster(Configuration conf) throws Exception {
  445. if (nameNode == null) {
  446. throw new IllegalStateException("Attempting to finalize "
  447. + "Namenode but it is not running");
  448. }
  449. ToolRunner.run(new DFSAdmin(conf), new String[] {"-finalizeUpgrade"});
  450. }
  451. /**
  452. * Gets the started NameNode. May be null.
  453. */
  454. public NameNode getNameNode() {
  455. return nameNode;
  456. }
  457. /**
  458. * Gets a list of the started DataNodes. May be empty.
  459. */
  460. public ArrayList<DataNode> getDataNodes() {
  461. ArrayList<DataNode> list = new ArrayList<DataNode>();
  462. for (int i = 0; i < dataNodes.size(); i++) {
  463. DataNode node = dataNodes.get(i).datanode;
  464. list.add(node);
  465. }
  466. return list;
  467. }
  468. /** @return the datanode having the ipc server listen port */
  469. DataNode getDataNode(int ipcPort) {
  470. for(DataNode dn : getDataNodes()) {
  471. if (dn.ipcServer.getListenerAddress().getPort() == ipcPort) {
  472. return dn;
  473. }
  474. }
  475. return null;
  476. }
  477. /**
  478. * Gets the rpc port used by the NameNode, because the caller
  479. * supplied port is not necessarily the actual port used.
  480. */
  481. public int getNameNodePort() {
  482. return nameNode.getNameNodeAddress().getPort();
  483. }
  484. /**
  485. * Shut down the servers that are up.
  486. */
  487. public void shutdown() {
  488. System.out.println("Shutting down the Mini HDFS Cluster");
  489. shutdownDataNodes();
  490. if (nameNode != null) {
  491. nameNode.stop();
  492. nameNode.join();
  493. nameNode = null;
  494. }
  495. }
  496. /**
  497. * Shutdown all DataNodes started by this class. The NameNode
  498. * is left running so that new DataNodes may be started.
  499. */
  500. public void shutdownDataNodes() {
  501. for (int i = dataNodes.size()-1; i >= 0; i--) {
  502. System.out.println("Shutting down DataNode " + i);
  503. DataNode dn = dataNodes.remove(i).datanode;
  504. dn.shutdown();
  505. numDataNodes--;
  506. }
  507. }
  508. /*
  509. * Shutdown a particular datanode
  510. */
  511. boolean stopDataNode(int i) {
  512. if (i < 0 || i >= dataNodes.size()) {
  513. return false;
  514. }
  515. DataNode dn = dataNodes.remove(i).datanode;
  516. System.out.println("MiniDFSCluster Stopping DataNode " +
  517. dn.dnRegistration.getName() +
  518. " from a total of " + (dataNodes.size() + 1) +
  519. " datanodes.");
  520. dn.shutdown();
  521. numDataNodes--;
  522. return true;
  523. }
  524. /*
  525. * Restart a particular datanode
  526. */
  527. synchronized boolean restartDataNode(int i) throws IOException {
  528. if (i < 0 || i >= dataNodes.size()) {
  529. return false;
  530. }
  531. DataNodeProperties dnprop = dataNodes.remove(i);
  532. DataNode dn = dnprop.datanode;
  533. Configuration conf = dnprop.conf;
  534. String[] args = dnprop.dnArgs;
  535. System.out.println("MiniDFSCluster Restart DataNode " +
  536. dn.dnRegistration.getName() +
  537. " from a total of " + (dataNodes.size() + 1) +
  538. " datanodes.");
  539. dn.shutdown();
  540. // recreate new datanode with the same configuration as the one
  541. // that was stopped.
  542. Configuration newconf = new Configuration(conf); // save cloned config
  543. dataNodes.add(new DataNodeProperties(
  544. DataNode.createDataNode(args, conf),
  545. newconf, args));
  546. return true;
  547. }
  548. /*
  549. * Shutdown a datanode by name.
  550. */
  551. synchronized boolean stopDataNode(String name) {
  552. int i;
  553. for (i = 0; i < dataNodes.size(); i++) {
  554. DataNode dn = dataNodes.get(i).datanode;
  555. if (dn.dnRegistration.getName().equals(name)) {
  556. break;
  557. }
  558. }
  559. return stopDataNode(i);
  560. }
  561. /**
  562. * Returns true if the NameNode is running and is out of Safe Mode.
  563. */
  564. public boolean isClusterUp() {
  565. if (nameNode == null) {
  566. return false;
  567. }
  568. try {
  569. long[] sizes = nameNode.getStats();
  570. boolean isUp = false;
  571. synchronized (this) {
  572. isUp = (!nameNode.isInSafeMode() && sizes[0] != 0);
  573. }
  574. return isUp;
  575. } catch (IOException ie) {
  576. return false;
  577. }
  578. }
  579. /**
  580. * Returns true if there is at least one DataNode running.
  581. */
  582. public boolean isDataNodeUp() {
  583. if (dataNodes == null || dataNodes.size() == 0) {
  584. return false;
  585. }
  586. return true;
  587. }
  588. /**
  589. * Get a client handle to the DFS cluster.
  590. */
  591. public FileSystem getFileSystem() throws IOException {
  592. return FileSystem.get(conf);
  593. }
  594. /**
  595. * Get the directories where the namenode stores its state.
  596. */
  597. public Collection<File> getNameDirs() {
  598. return FSNamesystem.getNamespaceDirs(conf);
  599. }
  600. /**
  601. * Wait until the cluster is active and running.
  602. */
  603. public void waitActive() throws IOException {
  604. InetSocketAddress addr = new InetSocketAddress("localhost",
  605. getNameNodePort());
  606. DFSClient client = new DFSClient(addr, conf);
  607. DatanodeInfo[] dnInfos;
  608. // make sure all datanodes are alive
  609. while((dnInfos = client.datanodeReport(DatanodeReportType.LIVE)).length
  610. != numDataNodes) {
  611. try {
  612. Thread.sleep(500);
  613. } catch (Exception e) {
  614. }
  615. }
  616. int numResolved = 0;
  617. do {
  618. numResolved = 0;
  619. for (DatanodeInfo info : dnInfos) {
  620. if (!info.getNetworkLocation().equals(NetworkTopology.UNRESOLVED)) {
  621. numResolved++;
  622. } else {
  623. try {
  624. Thread.sleep(500);
  625. } catch (Exception e) {
  626. }
  627. dnInfos = client.datanodeReport(DatanodeReportType.LIVE);
  628. break;
  629. }
  630. }
  631. } while (numResolved != numDataNodes);
  632. client.close();
  633. }
  634. public void formatDataNodeDirs() throws IOException {
  635. base_dir = new File(System.getProperty("test.build.data"), "dfs/");
  636. data_dir = new File(base_dir, "data");
  637. if (data_dir.exists() && !FileUtil.fullyDelete(data_dir)) {
  638. throw new IOException("Cannot remove data directory: " + data_dir);
  639. }
  640. }
  641. /**
  642. *
  643. * @param dataNodeIndex - data node whose block report is desired - the index is same as for getDataNodes()
  644. * @return the block report for the specified data node
  645. */
  646. public Block[] getBlockReport(int dataNodeIndex) {
  647. if (dataNodeIndex < 0 || dataNodeIndex > dataNodes.size()) {
  648. throw new IndexOutOfBoundsException();
  649. }
  650. return dataNodes.get(dataNodeIndex).datanode.getFSDataset().getBlockReport();
  651. }
  652. /**
  653. *
  654. * @return block reports from all data nodes
  655. * Block[] is indexed in the same order as the list of datanodes returned by getDataNodes()
  656. */
  657. public Block[][] getAllBlockReports() {
  658. int numDataNodes = dataNodes.size();
  659. Block[][] result = new Block[numDataNodes][];
  660. for (int i = 0; i < numDataNodes; ++i) {
  661. result[i] = getBlockReport(i);
  662. }
  663. return result;
  664. }
  665. /**
  666. * This method is valid only if the the data nodes have simulated data
  667. * @param dataNodeIndex - data node i which to inject - the index is same as for getDataNodes()
  668. * @param blocksToInject - the blocks
  669. * @throws IOException
  670. * if not simulatedFSDataset
  671. * if any of blocks already exist in the data node
  672. *
  673. */
  674. public void injectBlocks(int dataNodeIndex, Block[] blocksToInject) throws IOException {
  675. if (dataNodeIndex < 0 || dataNodeIndex > dataNodes.size()) {
  676. throw new IndexOutOfBoundsException();
  677. }
  678. FSDatasetInterface dataSet = dataNodes.get(dataNodeIndex).datanode.getFSDataset();
  679. if (!(dataSet instanceof SimulatedFSDataset)) {
  680. throw new IOException("injectBlocks is valid only for SimilatedFSDataset");
  681. }
  682. SimulatedFSDataset sdataset = (SimulatedFSDataset) dataSet;
  683. sdataset.injectBlocks(blocksToInject);
  684. dataNodes.get(dataNodeIndex).datanode.scheduleBlockReport(0);
  685. }
  686. /**
  687. * This method is valid only if the the data nodes have simulated data
  688. * @param blocksToInject - blocksToInject[] is indexed in the same order as the list
  689. * of datanodes returned by getDataNodes()
  690. * @throws IOException
  691. * if not simulatedFSDataset
  692. * if any of blocks already exist in the data nodes
  693. * Note the rest of the blocks are not injected.
  694. */
  695. public void injectBlocks(Block[][] blocksToInject) throws IOException {
  696. if (blocksToInject.length > dataNodes.size()) {
  697. throw new IndexOutOfBoundsException();
  698. }
  699. for (int i = 0; i < blocksToInject.length; ++i) {
  700. injectBlocks(i, blocksToInject[i]);
  701. }
  702. }
  703. /**
  704. * Set the softLimit and hardLimit of client lease periods
  705. */
  706. void setLeasePeriod(long soft, long hard) {
  707. nameNode.namesystem.setLeasePeriod(soft, hard);
  708. }
  709. /**
  710. * Returns the current set of datanodes
  711. */
  712. DataNode[] listDataNodes() {
  713. DataNode[] list = new DataNode[dataNodes.size()];
  714. for (int i = 0; i < dataNodes.size(); i++) {
  715. list[i] = dataNodes.get(i).datanode;
  716. }
  717. return list;
  718. }
  719. }