HDDS-17261. Count disk I/O errors from the datanode data path per volume

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/utils/DiskCheckUtil.java

@@ -32,7 +32,11 @@
 import java.nio.file.NoSuchFileException;
 import java.nio.file.Path;
 import java.util.Arrays;
+import java.util.Collections;
+import java.util.IdentityHashMap;
+import java.util.List;
 import java.util.Random;
+import java.util.Set;
 import java.util.UUID;
 import org.apache.commons.io.IOUtils;
 import org.apache.ratis.util.FileUtils;
@@ -45,6 +49,17 @@
  */
 public final class DiskCheckUtil {
   public static final String LINUX_DISK_FULL_MESSAGE = "No space left on device";
+  /**
+   * Linux error messages, as reported by the JDK and RocksDB, for errno values that mean the device or file system
+   * under a volume is failing: EIO, EROFS, ENXIO and ENODEV, EUCLEAN and EREMOTEIO.
+   */
+  private static final List<String> LINUX_DISK_IO_ERROR_MESSAGES = Arrays.asList(
+      "Input/output error",
+      "Read-only file system",
+      "No such device",
+      "Structure needs cleaning",
+      "Remote I/O error");
+  private static final int MAX_CAUSE_CHAIN_DEPTH = 64;
   // For testing purposes, an alternate check implementation can be provided
   // to inject failures.
   private static DiskChecks impl = new DiskChecksImpl();
@@ -75,6 +90,27 @@ public static boolean checkReadWrite(File storageDir, File testFileDir,
     return impl.checkReadWrite(storageDir, testFileDir, numBytesToWrite);
   }
 
+  /**
+   * Returns true if the throwable or one of its causes indicates that the device or file system under a volume is
+   * failing, as opposed to errors like a missing file, a full disk or too many open files.
+   */
+  public static boolean isDiskIOError(Throwable throwable) {
+    Set<Throwable> visited = Collections.newSetFromMap(new IdentityHashMap<>());
+    int depth = 0;
+    for (Throwable cause = throwable;
+        cause != null && depth < MAX_CAUSE_CHAIN_DEPTH && visited.add(cause);
+        cause = cause.getCause(), depth++) {
+      if (cause instanceof SyncFailedException) {
+        return true;
+      }
+      String message = cause.getMessage();
+      if (message != null && LINUX_DISK_IO_ERROR_MESSAGES.stream().anyMatch(message::contains)) {
+        return true;
+      }
+    }
+    return false;
+  }
+
   /**
    * Defines operations that must be implemented by a class injecting
    * failures into this class. Default implementations return true so that

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/utils/HddsVolumeUtil.java

@@ -109,7 +109,7 @@ private static void loadVolume(HddsVolume volume, boolean readOnly,
     try {
       volume.loadDbStore(readOnly);
     } catch (IOException e) {
-      onFailure(volume);
+      onFailure(volume, e);
       if (logger != null) {
         logger.error("Load db store for HddsVolume {} failed",
             volume.getStorageDir().getAbsolutePath(), e);

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/utils/StorageVolumeUtil.java

@@ -59,6 +59,17 @@ public static void onFailure(StorageVolume volume) {
     }
   }
 
+  /**
+   * Same as {@link #onFailure(StorageVolume)}, but first records the error on the volume if it is a disk I/O error,
+   * so that disks failing on the data path can be identified even if the volume check itself passes.
+   */
+  public static void onFailure(StorageVolume volume, Throwable cause) {
+    if (volume instanceof HddsVolume && DiskCheckUtil.isDiskIOError(cause)) {
+      ((HddsVolume) volume).recordDataPathIOError(cause);
+    }
+    onFailure(volume);
+  }
+
   public static List<HddsVolume> getHddsVolumesList(
       List<StorageVolume> volumes) {
     return volumes.stream().

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/volume/HddsVolume.java

@@ -108,6 +108,8 @@ public class HddsVolume extends StorageVolume {
   private File deletedContainerDir;
   private final AtomicBoolean dbLoaded = new AtomicBoolean(false);
   private final AtomicBoolean dbLoadFailure = new AtomicBoolean(false);
+  // Time the last data path disk I/O error was logged, to log at most one every disk check min gap.
+  private final AtomicLong lastDataPathIOErrorLogTime = new AtomicLong(Long.MIN_VALUE);
 
   /**
    * Builder for HddsVolume.
@@ -189,6 +191,22 @@ public VolumeInfoMetrics getVolumeInfoStats() {
     return volumeInfoMetrics;
   }
 
+  /**
+   * Records a disk I/O error, such as EIO, hit while reading or writing container data or metadata on this volume.
+   * The volume state is not changed. The error is counted in {@link VolumeInfoMetrics#getNumDataPathIOErrors()} so
+   * that a disk failing reads of existing data can be identified even while the volume check keeps passing.
+   */
+  public void recordDataPathIOError(Throwable cause) {
+    volumeInfoMetrics.incNumDataPathIOErrors();
+    final long minGap = getDatanodeConfig() == null ? 0 : getDatanodeConfig().getDiskCheckMinGap().toMillis();
+    final long now = Time.monotonicNow();
+    final long last = lastDataPathIOErrorLogTime.get();
+    if ((last == Long.MIN_VALUE || now - last >= minGap) && lastDataPathIOErrorLogTime.compareAndSet(last, now)) {
+      LOG.warn("Volume {} hit a disk I/O error while reading or writing container data: {}. {} disk I/O errors on " +
+          "this volume since the datanode started.", this, cause, volumeInfoMetrics.getNumDataPathIOErrors());
+    }
+  }
+
   @Override
   protected StorageLocationReport.Builder reportBuilder() {
     StorageLocationReport.Builder builder = super.reportBuilder();

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/volume/VolumeInfoMetrics.java

@@ -87,6 +87,9 @@ public class VolumeInfoMetrics implements MetricsSource {
   @Metric("Number of scans skipped for the volume")
   private MutableCounterLong numScansSkipped;
 
+  @Metric("Number of disk I/O errors hit while reading or writing container data and metadata on the volume")
+  private MutableCounterLong numDataPathIOErrors;
+
   @Metric("Write requests allowed while usable space is between the reported (soft) and hard min-free-space thresholds")
   private MutableCounterLong numWriteRequestsInSoftBandMinFreeSpace;
 
@@ -200,6 +203,14 @@ public void incNumScans() {
     numScans.incr();
   }
 
+  public long getNumDataPathIOErrors() {
+    return numDataPathIOErrors.value();
+  }
+
+  public void incNumDataPathIOErrors() {
+    numDataPathIOErrors.incr();
+  }
+
   public long getNumScansSkipped() {
     return numScansSkipped.value();
   }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java

@@ -313,7 +313,7 @@ private void writeToContainerFile(File containerFile, boolean isCreate)
       }
 
     } catch (IOException ex) {
-      onFailure(containerData.getVolume());
+      onFailure(containerData.getVolume(), ex);
       final String op = tempContainerFile == null ? "create tmp file for "
           : (isCreate ? "create" : "update") + " container file ";
       throw new StorageContainerException(
@@ -353,7 +353,7 @@ public void delete() throws StorageContainerException {
       // Container will be removed from tmp directory under the volume.
       // On datanode shutdown/restart any partial artifacts left
       // will be wiped from volume's tmp directory.
-      onFailure(containerData.getVolume());
+      onFailure(containerData.getVolume(), ex);
       final String errMsg = "Failed to cleanup " + containerData;
       LOG.error(errMsg, ex);
       throw new StorageContainerException(errMsg, ex, CONTAINER_INTERNAL_ERROR);
@@ -514,7 +514,7 @@ private void compactDB() throws StorageContainerException {
       throw ex;
     } catch (IOException ex) {
       LOG.error("Error in DB compaction while closing container", ex);
-      onFailure(containerData.getVolume());
+      onFailure(containerData.getVolume(), ex);
       throw new StorageContainerException(ex, ERROR_IN_COMPACT_DB);
     }
   }
@@ -529,7 +529,7 @@ private void flushAndSyncDB() throws StorageContainerException {
       throw ex;
     } catch (IOException ex) {
       LOG.error("Error in DB sync while closing container", ex);
-      onFailure(containerData.getVolume());
+      onFailure(containerData.getVolume(), ex);
       throw new StorageContainerException(ex, ERROR_IN_DB_SYNC);
     }
   }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/helpers/BlockUtils.java

@@ -136,7 +136,7 @@ public static DBHandle getDB(KeyValueContainerData containerData,
             containerData.getSchemaVersion(), conf);
       }
     } catch (IOException ex) {
-      onFailure(containerData.getVolume());
+      onFailure(containerData.getVolume(), ex);
       String message = String.format("Error opening DB. Container:%s " +
           "ContainerPath:%s", containerData.getContainerID(), containerData
           .getDbFile().getPath());

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/helpers/ChunkUtils.java

@@ -154,7 +154,7 @@ private static void writeData(ChunkBuffer data, String filename,
       bytesWritten = writer.applyAsLong(data);
     } catch (UncheckedIOException e) {
       if (!(e.getCause() instanceof InterruptedIOException)) {
-        onFailure(volume);
+        onFailure(volume, e);
       }
       throw wrapInStorageContainerException(e.getCause());
     }
@@ -228,7 +228,7 @@ private static void readData(File file, long offset, long len,
          FileChannel channel = open(path, READ_OPTIONS, NO_ATTRIBUTES)) {
       bytesRead = readMethod.apply(channel);
     } catch (Exception e) {
-      onFailure(volume);
+      onFailure(volume, e);
       throw wrapInStorageContainerException(e);
     }
 

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/FilePerBlockStrategy.java

@@ -159,7 +159,7 @@ public void writeChunk(Container container, BlockID blockID, ChunkInfo info,
     try {
       channel = files.getChannel(chunkFile, doSyncWrite);
     } catch (IOException e) {
-      onFailure(volume);
+      onFailure(volume, e);
       throw e;
     }
 
@@ -290,7 +290,7 @@ public void finishWriteChunks(KeyValueContainer container,
       files.close(chunkFile);
       verifyChunkFileExists(chunkFile);
     } catch (IOException e) {
-      onFailure(container.getContainerData().getVolume());
+      onFailure(container.getContainerData().getVolume(), e);
       throw e;
     }
   }
@@ -306,7 +306,7 @@ public void finalizeWriteChunk(KeyValueContainer container,
         }
         verifyChunkFileExists(chunkFile);
       } catch (IOException e) {
-        onFailure(container.getContainerData().getVolume());
+        onFailure(container.getContainerData().getVolume(), e);
         throw e;
       }
     }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/StreamDataChannelBase.java

@@ -75,16 +75,16 @@ private FileChannel getChannel() {
     return randomAccessFile.getChannel();
   }
 
-  protected void checkVolume() {
-    onFailure(containerData.getVolume());
+  protected void checkVolume(IOException e) {
+    onFailure(containerData.getVolume(), e);
   }
 
   @Override
   public final void force(boolean metadata) throws IOException {
     try {
       getChannel().force(metadata);
     } catch (IOException e) {
-      checkVolume();
+      checkVolume(e);
       throw e;
     }
   }
@@ -144,7 +144,7 @@ public void close() throws IOException {
     try {
       randomAccessFile.close();
     } catch (IOException e) {
-      checkVolume();
+      checkVolume(e);
       throw e;
     }
   }
@@ -158,7 +158,7 @@ final int writeFileChannel(ByteBuffer src) throws IOException {
       metrics.incContainerOpsLatencies(getType(), Time.monotonicNowNanos() - startTime);
       return writeBytes;
     } catch (IOException e) {
-      checkVolume();
+      checkVolume(e);
       throw e;
     }
   }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/ContainerScanHelper.java

@@ -26,6 +26,7 @@
 import org.apache.hadoop.ozone.container.common.impl.ContainerData;
 import org.apache.hadoop.ozone.container.common.interfaces.Container;
 import org.apache.hadoop.ozone.container.common.interfaces.ScanResult;
+import org.apache.hadoop.ozone.container.common.utils.DiskCheckUtil;
 import org.apache.hadoop.ozone.container.common.utils.StorageVolumeUtil;
 import org.apache.hadoop.ozone.container.common.volume.HddsVolume;
 import org.slf4j.Logger;
@@ -149,6 +150,8 @@ public void handleUnhealthyScanResult(ContainerData containerData, ScanResult re
       log.debug("Complete list of errors detected while scanning container {}:\n{}", containerID, allErrorString);
     }
 
+    recordDiskIOError(containerData, result);
+
     // Only increment the number of unhealthy containers if the container was not already unhealthy.
     // TODO HDDS-11593: Scanner counters will start from zero
     //  at the beginning of each run, so this will need to be incremented for every unhealthy container seen
@@ -161,6 +164,22 @@ public void handleUnhealthyScanResult(ContainerData containerData, ScanResult re
     }
   }
 
+  /**
+   * Counts a disk I/O error found by the scan against the container's volume, so that a disk with bad sectors can be
+   * identified even if the volume check itself passes.
+   */
+  private void recordDiskIOError(ContainerData containerData, ScanResult result) {
+    HddsVolume volume = containerData.getVolume();
+    if (volume == null || volume.isFailed()) {
+      return;
+    }
+    result.getErrors().stream()
+        .map(ContainerScanError::getException)
+        .filter(DiskCheckUtil::isDiskIOError)
+        .findFirst()
+        .ifPresent(volume::recordDataPathIOError);
+  }
+
   public void triggerVolumeScan(ContainerData containerData) {
     HddsVolume volume = containerData.getVolume();
     if (volume != null && !volume.isFailed()) {

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/utils/TestDiskCheckUtil.java

@@ -17,6 +17,7 @@
 
 package org.apache.hadoop.ozone.container.common.utils;
 
+import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.IO_EXCEPTION;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -26,9 +27,15 @@
 import static org.mockito.Mockito.mockStatic;
 
 import java.io.File;
+import java.io.IOException;
+import java.io.SyncFailedException;
+import java.io.UncheckedIOException;
+import java.nio.file.AccessDeniedException;
 import java.nio.file.FileSystemException;
+import java.nio.file.NoSuchFileException;
 import java.nio.file.OpenOption;
 import java.nio.file.Path;
+import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
 import org.apache.ratis.util.FileUtils;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
@@ -111,4 +118,26 @@ public void testCheckReadWriteDiskFull() {
       assertTestFileDeleted();
     }
   }
+
+  @Test
+  public void testIsDiskIOError() {
+    assertTrue(DiskCheckUtil.isDiskIOError(new IOException("Input/output error")));
+    assertTrue(DiskCheckUtil.isDiskIOError(new FileSystemException("/data/chunk", null, "Read-only file system")));
+    assertTrue(DiskCheckUtil.isDiskIOError(new IOException("No such device or address")));
+    assertTrue(DiskCheckUtil.isDiskIOError(new IOException("Structure needs cleaning")));
+    assertTrue(DiskCheckUtil.isDiskIOError(new SyncFailedException("sync failed")));
+    assertTrue(DiskCheckUtil.isDiskIOError(
+        new IOException("IO error: While appending to file: /data/db/000123.log: Input/output error")));
+    // Disk I/O errors wrapped by the data path.
+    assertTrue(DiskCheckUtil.isDiskIOError(new UncheckedIOException(new IOException("Input/output error"))));
+    assertTrue(DiskCheckUtil.isDiskIOError(
+        new StorageContainerException(new IOException("Input/output error"), IO_EXCEPTION)));
+
+    assertFalse(DiskCheckUtil.isDiskIOError(null));
+    assertFalse(DiskCheckUtil.isDiskIOError(new NoSuchFileException("/data/chunk")));
+    assertFalse(DiskCheckUtil.isDiskIOError(new AccessDeniedException("/data/chunk")));
+    assertFalse(DiskCheckUtil.isDiskIOError(new IOException(DiskCheckUtil.LINUX_DISK_FULL_MESSAGE)));
+    assertFalse(DiskCheckUtil.isDiskIOError(new FileSystemException("/data/chunk", null, "Too many open files")));
+    assertFalse(DiskCheckUtil.isDiskIOError(new IOException("Checksum mismatch")));
+  }
 }

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestHddsVolume.java

@@ -28,7 +28,9 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.nio.file.Files;
+import java.nio.file.NoSuchFileException;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.nio.file.attribute.PosixFilePermission;
@@ -147,6 +149,23 @@ public void testHddsVolumeInitialization() throws Exception {
     volume.shutdown();
   }
 
+  @Test
+  public void testDataPathIOErrorsCounted() throws Exception {
+    HddsVolume volume = volumeBuilder.build();
+    volume.format(CLUSTER_ID);
+
+    StorageVolumeUtil.onFailure(volume, new IOException("Input/output error"));
+    StorageVolumeUtil.onFailure(volume, new UncheckedIOException(new IOException("Read-only file system")));
+    // Errors that do not point at a failing disk are not counted.
+    StorageVolumeUtil.onFailure(volume, new NoSuchFileException("/data/hdds/chunk"));
+    StorageVolumeUtil.onFailure(volume, new IOException("No space left on device"));
+
+    assertEquals(2, volume.getVolumeInfoStats().getNumDataPathIOErrors());
+    // Counting errors does not change the volume state.
+    assertEquals(HddsVolume.VolumeState.NORMAL, volume.getStorageState());
+    volume.shutdown();
+  }
+
   @Test
   public void testCreateVolumeTmpDirs() throws Exception {
     // Set up volume.