HDDS-8782. Improve Volume Scanner Health checks.

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

@@ -62,7 +62,6 @@
 import org.apache.hadoop.ozone.container.common.volume.HddsVolume;
 import org.apache.hadoop.ozone.container.common.volume.MutableVolumeSet;
 import org.apache.hadoop.ozone.container.common.volume.StorageVolume;
-import org.apache.hadoop.ozone.container.keyvalue.helpers.KeyValueContainerUtil;
 import org.apache.hadoop.ozone.util.OzoneNetUtils;
 import org.apache.hadoop.ozone.util.ShutdownHookManager;
 import org.apache.hadoop.security.SecurityUtil;
@@ -130,25 +129,6 @@ public HddsDatanodeService(boolean printBanner, String[] args) {
     this.args = args != null ? Arrays.copyOf(args, args.length) : null;
   }
 
-  private void cleanTmpDir() {
-    if (datanodeStateMachine != null) {
-      MutableVolumeSet volumeSet =
-          datanodeStateMachine.getContainer().getVolumeSet();
-      for (StorageVolume volume : volumeSet.getVolumesList()) {
-        if (volume instanceof HddsVolume) {
-          HddsVolume hddsVolume = (HddsVolume) volume;
-          try {
-            KeyValueContainerUtil.ContainerDeleteDirectory
-                .cleanTmpDir(hddsVolume);
-          } catch (IOException ex) {
-            LOG.error("Error while cleaning tmp delete directory " +
-                "under {}", hddsVolume.getWorkingDir(), ex);
-          }
-        }
-      }
-    }
-  }
-
   /**
    * Create a Datanode instance based on the supplied command-line arguments.
    * <p>
@@ -574,8 +554,6 @@ public void terminateDatanode() {
   @Override
   public void stop() {
     if (!isStopped.getAndSet(true)) {
-      // Clean <HddsVolume>/tmp/container_delete_service dir.
-      cleanTmpDir();
       if (plugins != null) {
         for (ServicePlugin plugin : plugins) {
           try {

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeConfiguration.java

@@ -44,6 +44,10 @@ public class DatanodeConfiguration {
       "hdds.datanode.periodic.disk.check.interval.minutes";
   public static final String FAILED_DATA_VOLUMES_TOLERATED_KEY =
       "hdds.datanode.failed.data.volumes.tolerated";
+  public static final String CONSECUTIVE_VOLUME_IO_FAILURES_TOLERATED_KEY =
+      "hdds.datanode.consecutive.volume.io.failures.tolerated";
+  public static final String VOLUME_HEALTH_CHECK_FILE_SIZE_KEY =
+      "hdds.datanode.volume.health.check.file.size";
   public static final String FAILED_METADATA_VOLUMES_TOLERATED_KEY =
       "hdds.datanode.failed.metadata.volumes.tolerated";
   public static final String FAILED_DB_VOLUMES_TOLERATED_KEY =
@@ -64,6 +68,10 @@ public class DatanodeConfiguration {
 
   static final int FAILED_VOLUMES_TOLERATED_DEFAULT = -1;
 
+  public static final int CONSECUTIVE_VOLUME_IO_FAILURES_TOLERATED_DEFAULT = 3;
+
+  public static final int VOLUME_HEALTH_CHECK_FILE_SIZE_DEFAULT = 100;
+
   static final boolean WAIT_ON_ALL_FOLLOWERS_DEFAULT = false;
 
   static final long DISK_CHECK_MIN_GAP_DEFAULT =
@@ -269,6 +277,29 @@ public void setBlockDeletionLimit(int limit) {
   )
   private int failedDbVolumesTolerated = FAILED_VOLUMES_TOLERATED_DEFAULT;
 
+  @Config(key = "consecutive.volume.io.failures.tolerated",
+      defaultValue = "3",
+      type = ConfigType.INT,
+      tags = { DATANODE },
+      description = "The number of consecutive disk health checks that can " +
+          "fail due to a read/write error before the volume is considered " +
+          "failed."
+  )
+  private int consecutiveVolumeIOFailuresTolerated =
+      CONSECUTIVE_VOLUME_IO_FAILURES_TOLERATED_DEFAULT;
+
+  @Config(key = "volume.health.check.file.size",
+      defaultValue = "100B",
+      type = ConfigType.SIZE,
+      tags = { DATANODE },
+      description = "The size of the temporary file that will be synced to " +
+          "the disk and " +
+          "read back to assess its health. The contents of the " +
+          "file will be stored in memory during the duration of the check."
+  )
+  private int volumeHealthCheckFileSize =
+      VOLUME_HEALTH_CHECK_FILE_SIZE_DEFAULT;
+
   @Config(key = "disk.check.min.gap",
       defaultValue = "15m",
       type = ConfigType.TIME,
@@ -451,6 +482,24 @@ public void validate() {
       failedDbVolumesTolerated = FAILED_VOLUMES_TOLERATED_DEFAULT;
     }
 
+    if (consecutiveVolumeIOFailuresTolerated < 1) {
+      LOG.warn(CONSECUTIVE_VOLUME_IO_FAILURES_TOLERATED_KEY +
+              "must be at least 1 and was set to {}. Defaulting to {}",
+          consecutiveVolumeIOFailuresTolerated,
+          CONSECUTIVE_VOLUME_IO_FAILURES_TOLERATED_DEFAULT);
+      consecutiveVolumeIOFailuresTolerated =
+          CONSECUTIVE_VOLUME_IO_FAILURES_TOLERATED_DEFAULT;
+    }
+
+    if (volumeHealthCheckFileSize < 1) {
+      LOG.warn(VOLUME_HEALTH_CHECK_FILE_SIZE_KEY +
+              "must be at least 1 byte and was set to {}. Defaulting to {}",
+          volumeHealthCheckFileSize,
+          VOLUME_HEALTH_CHECK_FILE_SIZE_DEFAULT);
+      volumeHealthCheckFileSize =
+          VOLUME_HEALTH_CHECK_FILE_SIZE_DEFAULT;
+    }
+
     if (diskCheckMinGap < 0) {
       LOG.warn(DISK_CHECK_MIN_GAP_KEY +
               " must be greater than zero and was set to {}. Defaulting to {}",
@@ -487,7 +536,6 @@ public void validate() {
       rocksdbDeleteObsoleteFilesPeriod =
           ROCKSDB_DELETE_OBSOLETE_FILES_PERIOD_MICRO_SECONDS_DEFAULT;
     }
-
   }
 
   public void setContainerDeleteThreads(int containerDeleteThreads) {
@@ -531,6 +579,22 @@ public void setFailedDbVolumesTolerated(int failedVolumesTolerated) {
     this.failedDbVolumesTolerated = failedVolumesTolerated;
   }
 
+  public int getConsecutiveVolumeIOFailuresTolerated() {
+    return consecutiveVolumeIOFailuresTolerated;
+  }
+
+  public void setConsecutiveVolumeIOFailuresTolerated(int failuresToTolerate) {
+    this.consecutiveVolumeIOFailuresTolerated = failuresToTolerate;
+  }
+
+  public int getVolumeHealthCheckFileSize() {
+    return volumeHealthCheckFileSize;
+  }
+
+  public void getVolumeHealthCheckFileSize(int fileSizeBytes) {
+    this.volumeHealthCheckFileSize = fileSizeBytes;
+  }
+
   public Duration getDiskCheckMinGap() {
     return Duration.ofMillis(diskCheckMinGap);
   }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/VersionEndpointTask.java

@@ -24,10 +24,8 @@
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.hadoop.ozone.container.common.statemachine.EndpointStateMachine;
 import org.apache.hadoop.ozone.container.common.utils.StorageVolumeUtil;
-import org.apache.hadoop.ozone.container.common.volume.HddsVolume;
 import org.apache.hadoop.ozone.container.common.volume.MutableVolumeSet;
 import org.apache.hadoop.ozone.container.common.volume.StorageVolume;
-import org.apache.hadoop.ozone.container.keyvalue.helpers.KeyValueContainerUtil;
 import org.apache.hadoop.ozone.container.ozoneimpl.OzoneContainer;
 import org.apache.hadoop.ozone.protocol.VersionResponse;
 import org.apache.hadoop.util.DiskChecker.DiskOutOfSpaceException;
@@ -124,20 +122,7 @@ private void checkVolumeSet(MutableVolumeSet volumeSet,
         boolean result = StorageVolumeUtil.checkVolume(volume,
             scmId, clusterId, configuration, LOG,
             ozoneContainer.getDbVolumeSet());
-
-        if (result) {
-          // Clean <HddsVolume>/tmp/container_delete_service dir.
-          if (volume instanceof HddsVolume) {
-            HddsVolume hddsVolume = (HddsVolume) volume;
-            try {
-              KeyValueContainerUtil.ContainerDeleteDirectory
-                  .cleanTmpDir(hddsVolume);
-            } catch (IOException ex) {
-              LOG.error("Error while cleaning tmp delete directory " +
-                  "under {}", hddsVolume.getWorkingDir(), ex);
-            }
-          }
-        } else {
+        if (!result) {
           volumeSet.failVolume(volume.getStorageDir().getPath());
         }
       }

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

@@ -0,0 +1,199 @@
+/*
+ * 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.ozone.container.common.utils;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.SyncFailedException;
+import java.util.Arrays;
+import java.util.Random;
+import java.util.UUID;
+
+/**
+ * Utility class that supports checking disk health when provided a directory
+ * where the disk is mounted.
+ */
+public final class DiskCheckUtil {
+  private DiskCheckUtil() { }
+
+  // For testing purposes, an alternate check implementation can be provided
+  // to inject failures.
+  private static DiskChecks impl = new DiskChecksImpl();
+
+  @VisibleForTesting
+  public static void setTestImpl(DiskChecks diskChecks) {
+    impl = diskChecks;
+  }
+
+  @VisibleForTesting
+  public static void clearTestImpl() {
+    impl = new DiskChecksImpl();
+  }
+
+  public static boolean checkExistence(File storageDir) {
+    return impl.checkExistence(storageDir);
+  }
+
+  public static boolean checkPermissions(File storageDir) {
+    return impl.checkPermissions(storageDir);
+  }
+
+  public static boolean checkReadWrite(File storageDir, File testFileDir,
+      int numBytesToWrite) {
+    return impl.checkReadWrite(storageDir, testFileDir, numBytesToWrite);
+  }
+
+  /**
+   * Defines operations that must be implemented by a class injecting
+   * failures into this class. Default implementations return true so that
+   * tests only need to override methods for the failures they want to test.
+   */
+  public interface DiskChecks {
+    default boolean checkExistence(File storageDir) {
+      return true;
+    }
+    default boolean checkPermissions(File storageDir) {
+      return true;
+    }
+    default boolean checkReadWrite(File storageDir, File testFileDir,
+                            int numBytesToWrite) {
+      return true;
+    }
+  }
+
+  /**
+   * The default implementation of DiskCheck that production code will use
+   * for disk checking.
+   */
+  private static class DiskChecksImpl implements DiskChecks {
+
+    private static final Logger LOG =
+        LoggerFactory.getLogger(DiskCheckUtil.class);
+
+    @Override
+    public boolean checkExistence(File diskDir) {
+      if (!diskDir.exists()) {
+        logError(diskDir, "Directory does not exist.");
+        return false;
+      }
+      return true;
+    }
+
+    @Override
+    public boolean checkPermissions(File storageDir) {
+      // Check all permissions on the volume. If there are multiple permission
+      // errors, count it as one failure so the admin can fix them all at once.
+      boolean permissionsCorrect = true;
+      if (!storageDir.canRead()) {
+        logError(storageDir,
+            "Datanode does not have read permission on volume.");
+        permissionsCorrect = false;
+      }
+      if (!storageDir.canWrite()) {
+        logError(storageDir,
+            "Datanode does not have write permission on volume.");
+        permissionsCorrect = false;
+      }
+      if (!storageDir.canExecute()) {
+        logError(storageDir, "Datanode does not have execute" +
+            "permission on volume.");
+        permissionsCorrect = false;
+      }
+
+      return permissionsCorrect;
+    }
+
+    @Override
+    public boolean checkReadWrite(File storageDir,
+        File testFileDir, int numBytesToWrite) {
+      File testFile = new File(testFileDir, "disk-check-" + UUID.randomUUID());
+      byte[] writtenBytes = new byte[numBytesToWrite];
+      new Random().nextBytes(writtenBytes);
+      try (FileOutputStream fos = new FileOutputStream(testFile)) {
+        fos.write(writtenBytes);
+        fos.getFD().sync();
+      } catch (FileNotFoundException notFoundEx) {
+        logError(storageDir, String.format("Could not find file %s for " +
+            "volume check.", testFile), notFoundEx);
+        return false;
+      } catch (SyncFailedException syncEx) {
+        logError(storageDir, String.format("Could sync file %s to disk.",
+            testFile), syncEx);
+        return false;
+      } catch (IOException ioEx) {
+        logError(storageDir, String.format("Could not write file %s " +
+            "for volume check.", testFile), ioEx);
+        return false;
+      }
+
+      // Read data back from the test file.
+      byte[] readBytes = new byte[numBytesToWrite];
+      try (FileInputStream fis = new FileInputStream(testFile)) {
+        int numBytesRead = fis.read(readBytes);
+        if (numBytesRead != numBytesToWrite) {
+          logError(storageDir, String.format("%d bytes written to file %s " +
+                  "but %d bytes were read back.", numBytesToWrite, testFile,
+              numBytesRead));
+          return false;
+        }
+      } catch (FileNotFoundException notFoundEx) {
+        logError(storageDir, String.format("Could not find file %s " +
+            "for volume check.", testFile), notFoundEx);
+        return false;
+      } catch (IOException ioEx) {
+        logError(storageDir, String.format("Could not read file %s " +
+            "for volume check.", testFile), ioEx);
+        return false;
+      }
+
+      // Check that test file has the expected content.
+      if (!Arrays.equals(writtenBytes, readBytes)) {
+        logError(storageDir, String.format("%d Bytes read from file " +
+                "%s do not match the %d bytes that were written.",
+            writtenBytes.length, testFile, readBytes.length));
+        return false;
+      }
+
+      // Delete the file.
+      if (!testFile.delete()) {
+        logError(storageDir, String.format("Could not delete file %s " +
+            "for volume check.", testFile));
+        return false;
+      }
+
+      // If all checks passed, the volume is healthy.
+      return true;
+    }
+
+    private void logError(File storageDir, String message) {
+      LOG.error("Volume {} failed health check. {}", storageDir, message);
+    }
+
+    private void logError(File storageDir, String message, Exception ex) {
+      LOG.error("Volume {} failed health check. {}", storageDir, message, ex);
+    }
+  }
+}