You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by su...@apache.org on 2017/03/11 02:22:34 UTC
[14/50] [abbrv] hadoop git commit: HDFS-11340. DataNode reconfigure
for disks doesn't remove the failed volumes. (Manoj Govindassamy via lei)
HDFS-11340. DataNode reconfigure for disks doesn't remove the failed volumes. (Manoj Govindassamy via lei)
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/6d356b6b
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/6d356b6b
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/6d356b6b
Branch: refs/heads/YARN-2915
Commit: 6d356b6b4d8ccb32397cacfb5d0357b21f6035fc
Parents: 9649c27
Author: Lei Xu <le...@apache.org>
Authored: Fri Mar 10 14:36:51 2017 -0800
Committer: Lei Xu <le...@apache.org>
Committed: Fri Mar 10 14:37:13 2017 -0800
----------------------------------------------------------------------
.../hadoop/hdfs/server/datanode/DataNode.java | 73 +++++++++---
.../datanode/fsdataset/impl/FsDatasetImpl.java | 14 ++-
.../datanode/fsdataset/impl/FsVolumeList.java | 13 ++-
.../TestDataNodeVolumeFailureReporting.java | 116 +++++++++++++++++--
4 files changed, 184 insertions(+), 32 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/6d356b6b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
index 6f24858..5a82850 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
@@ -652,48 +652,84 @@ public class DataNode extends ReconfigurableBase
ChangedVolumes parseChangedVolumes(String newVolumes) throws IOException {
Configuration conf = new Configuration();
conf.set(DFS_DATANODE_DATA_DIR_KEY, newVolumes);
- List<StorageLocation> locations = getStorageLocations(conf);
+ List<StorageLocation> newStorageLocations = getStorageLocations(conf);
- if (locations.isEmpty()) {
+ if (newStorageLocations.isEmpty()) {
throw new IOException("No directory is specified.");
}
- // Use the existing StorageLocation to detect storage type changes.
- Map<String, StorageLocation> existingLocations = new HashMap<>();
+ // Use the existing storage locations from the current conf
+ // to detect new storage additions or removals.
+ Map<String, StorageLocation> existingStorageLocations = new HashMap<>();
for (StorageLocation loc : getStorageLocations(getConf())) {
- existingLocations.put(loc.getNormalizedUri().toString(), loc);
+ existingStorageLocations.put(loc.getNormalizedUri().toString(), loc);
}
ChangedVolumes results = new ChangedVolumes();
- results.newLocations.addAll(locations);
+ results.newLocations.addAll(newStorageLocations);
for (Iterator<Storage.StorageDirectory> it = storage.dirIterator();
it.hasNext(); ) {
Storage.StorageDirectory dir = it.next();
boolean found = false;
- for (Iterator<StorageLocation> sl = results.newLocations.iterator();
- sl.hasNext(); ) {
- StorageLocation location = sl.next();
- if (location.matchesStorageDirectory(dir)) {
- sl.remove();
- StorageLocation old = existingLocations.get(
- location.getNormalizedUri().toString());
- if (old != null &&
- old.getStorageType() != location.getStorageType()) {
+ for (Iterator<StorageLocation> newLocationItr =
+ results.newLocations.iterator(); newLocationItr.hasNext();) {
+ StorageLocation newLocation = newLocationItr.next();
+ if (newLocation.matchesStorageDirectory(dir)) {
+ StorageLocation oldLocation = existingStorageLocations.get(
+ newLocation.getNormalizedUri().toString());
+ if (oldLocation != null &&
+ oldLocation.getStorageType() != newLocation.getStorageType()) {
throw new IOException("Changing storage type is not allowed.");
}
- results.unchangedLocations.add(location);
+ // Update the unchanged locations as this location
+ // from the new conf is really not a new one.
+ newLocationItr.remove();
+ results.unchangedLocations.add(newLocation);
found = true;
break;
}
}
+ // New conf doesn't have the storage location which available in
+ // the current storage locations. Add to the deactivateLocations list.
if (!found) {
+ LOG.info("Deactivation request received for active volume: "
+ + dir.getRoot().toString());
results.deactivateLocations.add(
StorageLocation.parse(dir.getRoot().toString()));
}
}
+ // Use the failed storage locations from the current conf
+ // to detect removals in the new conf.
+ if (getFSDataset().getNumFailedVolumes() > 0) {
+ for (String failedStorageLocation : getFSDataset()
+ .getVolumeFailureSummary().getFailedStorageLocations()) {
+ boolean found = false;
+ for (Iterator<StorageLocation> newLocationItr =
+ results.newLocations.iterator(); newLocationItr.hasNext();) {
+ StorageLocation newLocation = newLocationItr.next();
+ if (newLocation.getNormalizedUri().toString().equals(
+ failedStorageLocation)) {
+ // The failed storage is being re-added. DataNode#refreshVolumes()
+ // will take care of re-assessing it.
+ found = true;
+ break;
+ }
+ }
+
+ // New conf doesn't have this failed storage location.
+ // Add to the deactivate locations list.
+ if (!found) {
+ LOG.info("Deactivation request received for failed volume: "
+ + failedStorageLocation);
+ results.deactivateLocations.add(StorageLocation.parse(
+ failedStorageLocation));
+ }
+ }
+ }
+
return results;
}
@@ -716,8 +752,9 @@ public class DataNode extends ReconfigurableBase
}
try {
- if (numOldDataDirs + changedVolumes.newLocations.size() -
- changedVolumes.deactivateLocations.size() <= 0) {
+ if (numOldDataDirs + getFSDataset().getNumFailedVolumes()
+ + changedVolumes.newLocations.size()
+ - changedVolumes.deactivateLocations.size() <= 0) {
throw new IOException("Attempt to remove all volumes.");
}
if (!changedVolumes.newLocations.isEmpty()) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/6d356b6b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsDatasetImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsDatasetImpl.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsDatasetImpl.java
index aff19ce..169e0e6 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsDatasetImpl.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsDatasetImpl.java
@@ -502,8 +502,10 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
*/
@Override
public void removeVolumes(
- Collection<StorageLocation> storageLocationsToRemove,
+ final Collection<StorageLocation> storageLocsToRemove,
boolean clearFailure) {
+ Collection<StorageLocation> storageLocationsToRemove =
+ new ArrayList<>(storageLocsToRemove);
Map<String, List<ReplicaInfo>> blkToInvalidate = new HashMap<>();
List<String> storageToRemove = new ArrayList<>();
try (AutoCloseableLock lock = datasetLock.acquire()) {
@@ -541,6 +543,16 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
}
storageToRemove.add(sd.getStorageUuid());
+ storageLocationsToRemove.remove(sdLocation);
+ }
+ }
+
+ // A reconfigure can remove the storage location which is already
+ // removed when the failure was detected by DataNode#checkDiskErrorAsync.
+ // Now, lets remove this from the failed volume list.
+ if (clearFailure) {
+ for (StorageLocation storageLocToRemove : storageLocationsToRemove) {
+ volumes.removeVolumeFailureInfo(storageLocToRemove);
}
}
setupAsyncLazyPersistThreads();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/6d356b6b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeList.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeList.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeList.java
index 64921d7..e7f0228 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeList.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeList.java
@@ -371,8 +371,15 @@ class FsVolumeList {
}
void addVolumeFailureInfo(VolumeFailureInfo volumeFailureInfo) {
- volumeFailureInfos.put(volumeFailureInfo.getFailedStorageLocation(),
- volumeFailureInfo);
+ // There could be redundant requests for adding the same failed
+ // volume because of repeated DataNode reconfigure with same list
+ // of volumes. Ignoring update on failed volume so as to preserve
+ // old failed capacity details in the map.
+ if (!volumeFailureInfos.containsKey(volumeFailureInfo
+ .getFailedStorageLocation())) {
+ volumeFailureInfos.put(volumeFailureInfo.getFailedStorageLocation(),
+ volumeFailureInfo);
+ }
}
private void addVolumeFailureInfo(FsVolumeImpl vol) {
@@ -382,7 +389,7 @@ class FsVolumeList {
vol.getCapacity()));
}
- private void removeVolumeFailureInfo(StorageLocation location) {
+ void removeVolumeFailureInfo(StorageLocation location) {
volumeFailureInfos.remove(location);
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/6d356b6b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailureReporting.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailureReporting.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailureReporting.java
index fbbc7f9..a3850ef 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailureReporting.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeVolumeFailureReporting.java
@@ -25,16 +25,22 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
import java.io.File;
+import java.lang.management.ManagementFactory;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.concurrent.TimeUnit;
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
+
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.conf.ReconfigurationException;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.DFSConfigKeys;
@@ -51,6 +57,7 @@ import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.test.GenericTestUtils;
import org.apache.log4j.Level;
import org.junit.After;
+import org.junit.Assert;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
@@ -406,8 +413,8 @@ public class TestDataNodeVolumeFailureReporting {
DataNodeTestUtils.triggerHeartbeat(dns.get(0));
DataNodeTestUtils.triggerHeartbeat(dns.get(1));
- checkFailuresAtDataNode(dns.get(0), 1, false, dn1Vol1.getAbsolutePath());
- checkFailuresAtDataNode(dns.get(1), 1, false, dn2Vol1.getAbsolutePath());
+ checkFailuresAtDataNode(dns.get(0), 1, true, dn1Vol1.getAbsolutePath());
+ checkFailuresAtDataNode(dns.get(1), 1, true, dn2Vol1.getAbsolutePath());
// Ensure we wait a sufficient amount of time.
assert (WAIT_FOR_HEARTBEATS * 10) > WAIT_FOR_DEATH;
@@ -415,9 +422,9 @@ public class TestDataNodeVolumeFailureReporting {
// The NN reports two volume failures again.
DFSTestUtil.waitForDatanodeStatus(dm, 3, 0, 2,
origCapacity - (1*dnCapacity), WAIT_FOR_HEARTBEATS);
- checkAggregateFailuresAtNameNode(false, 2);
- checkFailuresAtNameNode(dm, dns.get(0), false, dn1Vol1.getAbsolutePath());
- checkFailuresAtNameNode(dm, dns.get(1), false, dn2Vol1.getAbsolutePath());
+ checkAggregateFailuresAtNameNode(true, 2);
+ checkFailuresAtNameNode(dm, dns.get(0), true, dn1Vol1.getAbsolutePath());
+ checkFailuresAtNameNode(dm, dns.get(1), true, dn2Vol1.getAbsolutePath());
// Reconfigure a third time with the failed volumes. Afterwards, we expect
// the same volume failures to be reported. (No double-counting.)
@@ -427,8 +434,8 @@ public class TestDataNodeVolumeFailureReporting {
DataNodeTestUtils.triggerHeartbeat(dns.get(0));
DataNodeTestUtils.triggerHeartbeat(dns.get(1));
- checkFailuresAtDataNode(dns.get(0), 1, false, dn1Vol1.getAbsolutePath());
- checkFailuresAtDataNode(dns.get(1), 1, false, dn2Vol1.getAbsolutePath());
+ checkFailuresAtDataNode(dns.get(0), 1, true, dn1Vol1.getAbsolutePath());
+ checkFailuresAtDataNode(dns.get(1), 1, true, dn2Vol1.getAbsolutePath());
// Ensure we wait a sufficient amount of time.
assert (WAIT_FOR_HEARTBEATS * 10) > WAIT_FOR_DEATH;
@@ -436,9 +443,9 @@ public class TestDataNodeVolumeFailureReporting {
// The NN reports two volume failures again.
DFSTestUtil.waitForDatanodeStatus(dm, 3, 0, 2,
origCapacity - (1*dnCapacity), WAIT_FOR_HEARTBEATS);
- checkAggregateFailuresAtNameNode(false, 2);
- checkFailuresAtNameNode(dm, dns.get(0), false, dn1Vol1.getAbsolutePath());
- checkFailuresAtNameNode(dm, dns.get(1), false, dn2Vol1.getAbsolutePath());
+ checkAggregateFailuresAtNameNode(true, 2);
+ checkFailuresAtNameNode(dm, dns.get(0), true, dn1Vol1.getAbsolutePath());
+ checkFailuresAtNameNode(dm, dns.get(1), true, dn2Vol1.getAbsolutePath());
// Replace failed volume with healthy volume and run reconfigure DataNode.
// The failed volume information should be cleared.
@@ -515,6 +522,95 @@ public class TestDataNodeVolumeFailureReporting {
}
/**
+ * Verify DataNode NumFailedVolumes and FailedStorageLocations
+ * after hot swap out of failed volume.
+ */
+ @Test
+ public void testHotSwapOutFailedVolumeAndReporting()
+ throws Exception {
+ final File dn0Vol1 = new File(dataDir, "data" + (2 * 0 + 1));
+ final File dn0Vol2 = new File(dataDir, "data" + (2 * 0 + 2));
+ final DataNode dn0 = cluster.getDataNodes().get(0);
+ final String oldDataDirs = dn0.getConf().get(
+ DFSConfigKeys.DFS_DATANODE_DATA_DIR_KEY);
+
+ MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
+ ObjectName mxbeanName = new ObjectName(
+ "Hadoop:service=DataNode,name=FSDatasetState-" + dn0.getDatanodeUuid());
+ int numFailedVolumes = (int) mbs.getAttribute(mxbeanName,
+ "NumFailedVolumes");
+ Assert.assertEquals(dn0.getFSDataset().getNumFailedVolumes(),
+ numFailedVolumes);
+ checkFailuresAtDataNode(dn0, 0, false, new String[] {});
+
+ // Fail dn0Vol1 first.
+ // Verify NumFailedVolumes and FailedStorageLocations are empty.
+ DataNodeTestUtils.injectDataDirFailure(dn0Vol1);
+ DataNodeTestUtils.waitForDiskError(dn0,
+ DataNodeTestUtils.getVolume(dn0, dn0Vol1));
+ numFailedVolumes = (int) mbs.getAttribute(mxbeanName, "NumFailedVolumes");
+ Assert.assertEquals(1, numFailedVolumes);
+ Assert.assertEquals(dn0.getFSDataset().getNumFailedVolumes(),
+ numFailedVolumes);
+ checkFailuresAtDataNode(dn0, 1, true,
+ new String[] {dn0Vol1.getAbsolutePath()});
+
+ // Reconfigure disks without fixing the failed disk.
+ // Verify NumFailedVolumes and FailedStorageLocations haven't changed.
+ try {
+ dn0.reconfigurePropertyImpl(DFSConfigKeys.DFS_DATANODE_DATA_DIR_KEY,
+ oldDataDirs);
+ fail("Reconfigure with failed disk should throw exception.");
+ } catch (ReconfigurationException e) {
+ Assert.assertTrue("Reconfigure exception doesn't have expected path!",
+ e.getCause().getMessage().contains(dn0Vol1.getAbsolutePath()));
+ }
+ numFailedVolumes = (int) mbs.getAttribute(mxbeanName, "NumFailedVolumes");
+ Assert.assertEquals(1, numFailedVolumes);
+ Assert.assertEquals(dn0.getFSDataset().getNumFailedVolumes(),
+ numFailedVolumes);
+ checkFailuresAtDataNode(dn0, 1, true,
+ new String[] {dn0Vol1.getAbsolutePath()});
+
+ // Hot swap out the failed volume.
+ // Verify NumFailedVolumes and FailedStorageLocations are reset.
+ String dataDirs = dn0Vol2.getPath();
+ dn0.reconfigurePropertyImpl(DFSConfigKeys.DFS_DATANODE_DATA_DIR_KEY,
+ dataDirs);
+ numFailedVolumes = (int) mbs.getAttribute(mxbeanName, "NumFailedVolumes");
+ Assert.assertEquals(0, numFailedVolumes);
+ Assert.assertEquals(dn0.getFSDataset().getNumFailedVolumes(),
+ numFailedVolumes);
+ checkFailuresAtDataNode(dn0, 0, true, new String[] {});
+
+ // Fix failure volume dn0Vol1 and remount it back.
+ // Verify NumFailedVolumes and FailedStorageLocations are empty.
+ DataNodeTestUtils.restoreDataDirFromFailure(dn0Vol1);
+ dn0.reconfigurePropertyImpl(DFSConfigKeys.DFS_DATANODE_DATA_DIR_KEY,
+ oldDataDirs);
+ numFailedVolumes = (int) mbs.getAttribute(mxbeanName, "NumFailedVolumes");
+ Assert.assertEquals(0, numFailedVolumes);
+ Assert.assertEquals(dn0.getFSDataset().getNumFailedVolumes(),
+ numFailedVolumes);
+ checkFailuresAtDataNode(dn0, 0, true, new String[] {});
+
+ // Fail dn0Vol2.
+ // Verify NumFailedVolumes and FailedStorageLocations are updated.
+ DataNodeTestUtils.injectDataDirFailure(dn0Vol2);
+ DataNodeTestUtils.waitForDiskError(dn0,
+ DataNodeTestUtils.getVolume(dn0, dn0Vol2));
+ numFailedVolumes = (int) mbs.getAttribute(mxbeanName, "NumFailedVolumes");
+ Assert.assertEquals(1, numFailedVolumes);
+ Assert.assertEquals(dn0.getFSDataset().getNumFailedVolumes(),
+ numFailedVolumes);
+ checkFailuresAtDataNode(dn0, 1, true,
+ new String[] {dn0Vol2.getAbsolutePath()});
+
+ // Verify DataNode tolerating one disk failure.
+ assertTrue(dn0.shouldRun());
+ }
+
+ /**
* Checks the NameNode for correct values of aggregate counters tracking failed
* volumes across all DataNodes.
*
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org