Skip to content

HDFS-17936. Detect slow disks separately by storage type #8558

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
lfxy wants to merge 1 commit into
base: trunk
Choose a base branch
from
Open

HDFS-17936. Detect slow disks separately by storage type #8558

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
99 changes: 63 additions & 36 deletions ...dfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/metrics/DataNodeDiskMetrics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,18 +17,19 @@
*/
package org.apache.hadoop.hdfs.server.datanode.metrics;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.classification.VisibleForTesting;
import org.apache.hadoop.thirdparty.com.google.common.collect.ImmutableMap;
import org.apache.hadoop.thirdparty.com.google.common.collect.Maps;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.classification.VisibleForTesting;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.server.datanode.DataNode;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.DataNodeVolumeMetrics;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsDatasetSpi;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi;
import org.apache.hadoop.hdfs.server.protocol.SlowDiskReports.DiskOp;
import org.apache.hadoop.thirdparty.com.google.common.collect.ImmutableMap;
import org.apache.hadoop.thirdparty.com.google.common.collect.Maps;
import org.apache.hadoop.util.Daemon;
import org.apache.hadoop.util.Preconditions;
import org.slf4j.Logger;
Expand All @@ -37,6 +38,7 @@
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.EnumMap;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
Expand Down Expand Up @@ -111,9 +113,11 @@ private void startDiskOutlierDetectionThread() {
public void run() {
while (shouldRun) {
if (dn.getFSDataset() != null) {
Map<String, Double> metadataOpStats = Maps.newHashMap();
Map<String, Double> readIoStats = Maps.newHashMap();
Map<String, Double> writeIoStats = Maps.newHashMap();
Map<StorageType, Map<String, Double>> metadataOpStats =
new EnumMap<>(StorageType.class);
Map<StorageType, Map<String, Double>> readIoStats = new EnumMap<>(StorageType.class);
Map<StorageType, Map<String, Double>> writeIoStats = new EnumMap<>(StorageType.class);
Map<String, StorageType> diskToStorageType = new HashMap<>();
FsDatasetSpi.FsVolumeReferences fsVolumeReferences = null;
try {
fsVolumeReferences = dn.getFSDataset().getFsVolumeReferences();
Expand All @@ -123,11 +127,18 @@ public void run() {
FsVolumeSpi volume = volumeIterator.next();
DataNodeVolumeMetrics metrics = volume.getMetrics();
String volumeName = volume.getBaseURI().getPath();
StorageType storageType = volume.getStorageType();
if (storageType == null) {
storageType = StorageType.DEFAULT;
}

metadataOpStats.put(volumeName,
metrics.getMetadataOperationMean());
readIoStats.put(volumeName, metrics.getReadIoMean());
writeIoStats.put(volumeName, metrics.getWriteIoMean());
diskToStorageType.put(volumeName, storageType);
metadataOpStats.computeIfAbsent(storageType, k -> Maps.newHashMap())
.put(volumeName, metrics.getMetadataOperationMean());
readIoStats.computeIfAbsent(storageType, k -> Maps.newHashMap())
.put(volumeName, metrics.getReadIoMean());
writeIoStats.computeIfAbsent(storageType, k -> Maps.newHashMap())
.put(volumeName, metrics.getWriteIoMean());
}
} finally {
if (fsVolumeReferences != null) {
Expand All @@ -147,19 +158,30 @@ public void run() {
detectAndUpdateDiskOutliers(metadataOpStats, readIoStats,
writeIoStats);

// Sort the slow disks by latency and extract the top n by maxSlowDisksToExclude.
// Sort the slow disks by latency and extract the top n by maxSlowDisksToExclude
// within each StorageType group independently.
if (maxSlowDisksToExclude > 0) {
ArrayList<DiskLatency> diskLatencies = new ArrayList<>();
for (Map.Entry<String, Map<DiskOp, Double>> diskStats :
diskOutliersStats.entrySet()) {
diskLatencies.add(new DiskLatency(diskStats.getKey(), diskStats.getValue()));
Map<StorageType, ArrayList<DiskLatency>> slowDisksByType =
new EnumMap<>(StorageType.class);
for (Map.Entry<String, Map<DiskOp, Double>> diskStats : diskOutliersStats.entrySet()) {
String diskPath = diskStats.getKey();
StorageType st = diskToStorageType.get(diskPath);
if (st == null) {
st = StorageType.DEFAULT;
}
slowDisksByType.computeIfAbsent(st, k -> new ArrayList<>())
.add(new DiskLatency(diskPath, diskStats.getValue()));
}

Collections.sort(diskLatencies, (o1, o2)
-> Double.compare(o2.getMaxLatency(), o1.getMaxLatency()));

slowDisksToExclude = diskLatencies.stream().limit(maxSlowDisksToExclude)
.map(DiskLatency::getSlowDisk).collect(Collectors.toList());
List<String> excludeList = new ArrayList<>();
for (Map.Entry<StorageType, ArrayList<DiskLatency>> entry : slowDisksByType.entrySet()) {
ArrayList<DiskLatency> typeSlowDisks = entry.getValue();
Collections.sort(typeSlowDisks,
(o1, o2) -> Double.compare(o2.getMaxLatency(), o1.getMaxLatency()));
excludeList.addAll(typeSlowDisks.stream().limit(maxSlowDisksToExclude)
.map(DiskLatency::getSlowDisk).collect(Collectors.toList()));
}
slowDisksToExclude = excludeList;
}
}

Expand All @@ -175,30 +197,35 @@ public void run() {
slowDiskDetectionDaemon.start();
}

private void detectAndUpdateDiskOutliers(Map<String, Double> metadataOpStats,
Map<String, Double> readIoStats, Map<String, Double> writeIoStats) {
private void detectAndUpdateDiskOutliers(Map<StorageType, Map<String, Double>> metadataOpStats,
Map<StorageType, Map<String, Double>> readIoStats,
Map<StorageType, Map<String, Double>> writeIoStats) {
Map<String, Map<DiskOp, Double>> diskStats = Maps.newHashMap();

// Get MetadataOp Outliers
Map<String, Double> metadataOpOutliers = slowDiskDetector
.getOutliers(metadataOpStats);
for (Map.Entry<String, Double> entry : metadataOpOutliers.entrySet()) {
addDiskStat(diskStats, entry.getKey(), DiskOp.METADATA, entry.getValue());
for (Map.Entry<StorageType, Map<String, Double>> entry : metadataOpStats.entrySet()) {
Map<String, Double> metadataOpOutliers = slowDiskDetector.getOutliers(entry.getValue());
for (Map.Entry<String, Double> outlier : metadataOpOutliers.entrySet()) {
addDiskStat(diskStats, outlier.getKey(), DiskOp.METADATA, outlier.getValue());
}
}

// Get ReadIo Outliers
Map<String, Double> readIoOutliers = slowDiskDetector
.getOutliers(readIoStats);
for (Map.Entry<String, Double> entry : readIoOutliers.entrySet()) {
addDiskStat(diskStats, entry.getKey(), DiskOp.READ, entry.getValue());
for (Map.Entry<StorageType, Map<String, Double>> entry : readIoStats.entrySet()) {
Map<String, Double> readIoOutliers = slowDiskDetector.getOutliers(entry.getValue());
for (Map.Entry<String, Double> outlier : readIoOutliers.entrySet()) {
addDiskStat(diskStats, outlier.getKey(), DiskOp.READ, outlier.getValue());
}
}

// Get WriteIo Outliers
Map<String, Double> writeIoOutliers = slowDiskDetector
.getOutliers(writeIoStats);
for (Map.Entry<String, Double> entry : writeIoOutliers.entrySet()) {
addDiskStat(diskStats, entry.getKey(), DiskOp.WRITE, entry.getValue());
for (Map.Entry<StorageType, Map<String, Double>> entry : writeIoStats.entrySet()) {
Map<String, Double> writeIoOutliers = slowDiskDetector.getOutliers(entry.getValue());
for (Map.Entry<String, Double> outlier : writeIoOutliers.entrySet()) {
addDiskStat(diskStats, outlier.getKey(), DiskOp.WRITE, outlier.getValue());
}
}

if (overrideStatus) {
diskOutliersStats = diskStats;
LOG.debug("Updated disk outliers.");
Expand Down
114 changes: 114 additions & 0 deletions ...rg/apache/hadoop/hdfs/server/datanode/metrics/TestDataNodeOutlierDetectionViaMetrics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,9 +18,18 @@

package org.apache.hadoop.hdfs.server.datanode.metrics;

import java.net.URI;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Supplier;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.HdfsConfiguration;
import org.apache.hadoop.hdfs.server.datanode.DataNode;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.DataNodeVolumeMetrics;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsDatasetSpi;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi;
import org.apache.hadoop.hdfs.server.protocol.OutlierMetrics;
import org.apache.hadoop.metrics2.lib.MetricsTestHelper;
import org.apache.hadoop.test.GenericTestUtils;
Expand All @@ -35,8 +44,12 @@
import java.util.Random;
import java.util.concurrent.TimeUnit;

import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

/**
* Test that the {@link DataNodePeerMetrics} class is able to detect
Expand Down Expand Up @@ -156,4 +169,105 @@ public void injectSlowNodeSamples(
slowNodeName, SLOW_NODE_LATENCY_MS);
}
}

/**
* Verifies that slow disk detection is performed per StorageType group.
* Scenario 1: 1 SSD + 10 HDDs — SSD group skipped (size below minimum),
* 2 HDDs detected.
* Scenario 2: 6 SSDs + 10 HDDs — 2 SSDs and 2 HDDs
* each detected independently within their own group.
*/
@Test
public void testStorageTypeAwareSlowDiskDetection() throws Exception {
Configuration testConf = new HdfsConfiguration();
testConf.setLong(DFSConfigKeys.DFS_DATANODE_MIN_OUTLIER_DETECTION_DISKS_KEY, 5);
testConf.setInt(DFSConfigKeys.DFS_DATANODE_MAX_SLOWDISKS_TO_EXCLUDE_KEY, 1);

// Scenario 1: 1 SSD + 10 HDDs.
{
List<FsVolumeSpi> volumes = new ArrayList<>();
volumes.add(createMockDiskVolume("/ssd0/", StorageType.SSD, 5000.0));
for (int i = 0; i < 8; i++) {
volumes.add(createMockDiskVolume("/hdd" + i + "/", StorageType.DISK, 0.5));
}
volumes.add(createMockDiskVolume("/hdd8/", StorageType.DISK, 5000.0));
volumes.add(createMockDiskVolume("/hdd9/", StorageType.DISK, 6000.0));

DataNodeDiskMetrics diskMetrics = buildMetrics(testConf, volumes);
try {
GenericTestUtils.waitFor(() -> diskMetrics.getDiskOutliersStats().size() >= 2, 100, 10_000);
Map<String, ?> outliers = diskMetrics.getDiskOutliersStats();
assertFalse(outliers.containsKey("/ssd0/"), "SSD group too small, must not be flagged");
assertTrue(outliers.containsKey("/hdd8/"), "Slow HDD must be detected");
assertTrue(outliers.containsKey("/hdd9/"), "Slow HDD must be detected");
assertThat(diskMetrics.getSlowDisksToExclude()).hasSize(1);
} finally {
diskMetrics.shutdownAndWait();
}
}

// Scenario 2: 6 SSDs + 10 HDDs.
{
List<FsVolumeSpi> volumes = new ArrayList<>();
for (int i = 0; i < 4; i++) {
volumes.add(createMockDiskVolume("/ssd" + i + "/", StorageType.SSD, i + 1.0));
}
volumes.add(createMockDiskVolume("/ssd4/", StorageType.SSD, 5000.0));
volumes.add(createMockDiskVolume("/ssd5/", StorageType.SSD, 6000.0));
for (int i = 0; i < 8; i++) {
volumes.add(createMockDiskVolume("/hdd" + i + "/", StorageType.DISK, 0.5));
}
volumes.add(createMockDiskVolume("/hdd8/", StorageType.DISK, 5000.0));
volumes.add(createMockDiskVolume("/hdd9/", StorageType.DISK, 6000.0));

DataNodeDiskMetrics diskMetrics = buildMetrics(testConf, volumes);
try {
GenericTestUtils.waitFor(() -> diskMetrics.getDiskOutliersStats().size() >= 4, 100, 10_000);
Map<String, ?> outliers = diskMetrics.getDiskOutliersStats();
assertTrue(outliers.containsKey("/ssd4/"), "Slow SSD must be detected");
assertTrue(outliers.containsKey("/ssd5/"), "Slow SSD must be detected");
assertTrue(outliers.containsKey("/hdd8/"), "Slow HDD must be detected");
assertTrue(outliers.containsKey("/hdd9/"), "Slow HDD must be detected");
assertThat(outliers.keySet().stream().filter(k -> k.startsWith("/ssd")).count()).isEqualTo(
2);
assertThat(outliers.keySet().stream().filter(k -> k.startsWith("/hdd")).count()).isEqualTo(
2);
assertThat(diskMetrics.getSlowDisksToExclude()).hasSize(2);
} finally {
diskMetrics.shutdownAndWait();
}
}
}

private DataNodeDiskMetrics buildMetrics(Configuration conf, List<FsVolumeSpi> volumes) {
DataNode mockDn = mock(DataNode.class);
@SuppressWarnings({"unchecked", "rawtypes"})
FsDatasetSpi mockDataset = mock(FsDatasetSpi.class);
FsDatasetSpi.FsVolumeReferences mockRefs = mock(FsDatasetSpi.FsVolumeReferences.class);
when(mockDn.getFSDataset()).thenReturn(mockDataset);
when(mockDataset.getFsVolumeReferences()).thenReturn(mockRefs);
doAnswer(inv -> volumes.iterator()).when(mockRefs).iterator();
return new DataNodeDiskMetrics(mockDn, 100, conf);
}

/**
* Creates a mock volume; only metadata latency drives outlier detection.
*/
@SuppressWarnings("unchecked")
private FsVolumeSpi createMockDiskVolume(String path, StorageType storageType,
double metadataMs) {
FsVolumeSpi mockVolume = mock(FsVolumeSpi.class);
DataNodeVolumeMetrics mockMetrics = mock(DataNodeVolumeMetrics.class);
try {
when(mockVolume.getBaseURI()).thenReturn(new URI("file://" + path));
} catch (Exception e) {
throw new RuntimeException(e);
}
when(mockVolume.getStorageType()).thenReturn(storageType);
when(mockVolume.getMetrics()).thenReturn(mockMetrics);
when(mockMetrics.getMetadataOperationMean()).thenReturn(metadataMs);
when(mockMetrics.getReadIoMean()).thenReturn(0.0);
when(mockMetrics.getWriteIoMean()).thenReturn(0.0);
return mockVolume;
}
}
Loading