HDDS-8118. Fail container delete on non empty chunks dir

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ContainerMetrics.java

@@ -49,6 +49,10 @@ public class ContainerMetrics {
   public static final String STORAGE_CONTAINER_METRICS =
       "StorageContainerMetrics";
   @Metric private MutableCounterLong numOps;
+  @Metric private MutableCounterLong containerDeleteFailedNonEmptyDir;
+  @Metric private MutableCounterLong containerDeleteFailedBlockCountNotZero;
+  @Metric private MutableCounterLong containerForceDelete;
+
   private MutableCounterLong[] numOpsArray;
   private MutableCounterLong[] opsBytesArray;
   private MutableRate[] opsLatency;
@@ -125,4 +129,27 @@ public void incContainerBytesStats(ContainerProtos.Type type, long bytes) {
   public long getContainerBytesMetrics(ContainerProtos.Type type) {
     return opsBytesArray[type.ordinal()].value();
   }
+
+  public void incContainerDeleteFailedBlockCountNotZero() {
+    containerDeleteFailedBlockCountNotZero.incr();
+  }
+  public void incContainerDeleteFailedNonEmpty() {
+    containerDeleteFailedNonEmptyDir.incr();
+  }
+
+  public void incContainersForceDelete() {
+    containerForceDelete.incr();
+  }
+
+  public long getContainerDeleteFailedNonEmptyDir() {
+    return containerDeleteFailedNonEmptyDir.value();
+  }
+
+  public long getContainerDeleteFailedBlockCountNotZero() {
+    return containerDeleteFailedBlockCountNotZero.value();
+  }
+
+  public long getContainerForceDelete() {
+    return containerForceDelete.value();
+  }
 }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/Container.java

@@ -57,6 +57,13 @@ void create(VolumeSet volumeSet, VolumeChoosingPolicy volumeChoosingPolicy,
    */
   void delete() throws StorageContainerException;
 
+  /**
+   * Returns true if container is empty.
+   * @return true of container is empty
+   * @throws IOException if was unable to check container status.
+   */
+  boolean isEmpty() throws IOException;
+
   /**
    * Update the container.
    *

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

@@ -87,6 +87,13 @@ public class DatanodeConfiguration {
       ROCKSDB_DELETE_OBSOLETE_FILES_PERIOD_MICRO_SECONDS_KEY =
       "hdds.datanode.rocksdb.delete_obsolete_files_period";
 
+  public static final String
+      OZONE_DATANODE_CHECK_EMPTY_CONTAINER_ON_DISK_ON_DELETE =
+      "hdds.datanode.check.empty.container.delete";
+  public static final Boolean
+      OZONE_DATANODE_CHECK_EMPTY_CONTAINER_ON_DISK_ON_DELETE_DEFAULT =
+      true;
+
   /**
    * Number of threads per volume that Datanode will use for chunk read.
    */

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

@@ -308,6 +308,11 @@ public void delete() throws StorageContainerException {
     }
   }
 
+  @Override
+  public boolean isEmpty() throws IOException {
+    return KeyValueContainerUtil.noBlocksInContainer(containerData);
+  }
+
   @Override
   public void markContainerForClose() throws StorageContainerException {
     writeLock();

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

@@ -131,6 +131,7 @@ public class KeyValueHandler extends Handler {
   private final long maxContainerSize;
   private final Function<ByteBuffer, ByteString> byteBufferToByteString;
   private final boolean validateChunkChecksumData;
+  private boolean checkIfNoBlockFiles;
 
   // A striped lock that is held during container creation.
   private final Striped<Lock> containerCreationLocks;
@@ -155,6 +156,14 @@ public KeyValueHandler(ConfigurationSource config,
     } catch (Exception e) {
       throw new RuntimeException(e);
     }
+
+    checkIfNoBlockFiles =
+        conf.getBoolean(
+            DatanodeConfiguration.
+                OZONE_DATANODE_CHECK_EMPTY_CONTAINER_ON_DISK_ON_DELETE,
+        DatanodeConfiguration.
+            OZONE_DATANODE_CHECK_EMPTY_CONTAINER_ON_DISK_ON_DELETE_DEFAULT);
+
     maxContainerSize = (long) config.getStorageSize(
         ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE,
         ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE_DEFAULT, StorageUnit.BYTES);
@@ -1231,19 +1240,45 @@ private void deleteInternal(Container container, boolean force)
         }
         // Safety check that the container is empty.
         // If the container is not empty, it should not be deleted unless the
-        // container is beinf forcefully deleted (which happens when
+        // container is being forcefully deleted (which happens when
         // container is unhealthy or over-replicated).
         if (container.getContainerData().getBlockCount() != 0) {
+          metrics.incContainerDeleteFailedBlockCountNotZero();
           LOG.error("Received container deletion command for container {} but" +
-              " the container is not empty.",
-              container.getContainerData().getContainerID());
+              " the container is not empty with blockCount {}",
+              container.getContainerData().getContainerID(),
+              container.getContainerData().getBlockCount());
           throw new StorageContainerException("Non-force deletion of " +
               "non-empty container is not allowed.",
               DELETE_ON_NON_EMPTY_CONTAINER);
         }
+
+        if (checkIfNoBlockFiles && !container.isEmpty()) {
+          metrics.incContainerDeleteFailedNonEmpty();
+          LOG.error("Received container deletion command for container {} but" +
+                  " the container is not empty",
+              container.getContainerData().getContainerID());
+          throw new StorageContainerException("Non-force deletion of " +
+              "non-empty container:" +
+              container.getContainerData().getContainerID() +
+              " is not allowed.",
+              DELETE_ON_NON_EMPTY_CONTAINER);
+        }
+      } else {
+        metrics.incContainersForceDelete();
       }
       long containerId = container.getContainerData().getContainerID();
       containerSet.removeContainer(containerId);
+    } catch (StorageContainerException e) {
+      throw e;
+    } catch (IOException e) {
+      // All other IO Exceptions should be treated as if the container is not
+      // empty as a defensive check.
+      LOG.error("Could not determine if the container {} is empty",
+          container.getContainerData().getContainerID(), e);
+      throw new StorageContainerException("Could not determine if container "
+          + container.getContainerData().getContainerID() +
+          " is empty", DELETE_ON_NON_EMPTY_CONTAINER);
     } finally {
       container.writeUnlock();
     }

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

@@ -19,6 +19,7 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.nio.file.DirectoryStream;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
@@ -159,6 +160,26 @@ public static void removeContainer(KeyValueContainerData containerData,
     FileUtils.deleteDirectory(containerMetaDataPath.getParentFile());
   }
 
+  /**
+   * Returns if there are no blocks in the container.
+   * @param containerData Container to check
+   * @param conf configuration
+   * @return true if the directory containing blocks is empty
+   * @throws IOException
+   */
+  public static boolean noBlocksInContainer(KeyValueContainerData
+                                                containerData)
+      throws IOException {
+    Preconditions.checkNotNull(containerData);
+    File chunksPath = new File(containerData.getChunksPath());
+    Preconditions.checkArgument(chunksPath.isDirectory());
+
+    try (DirectoryStream<Path> dir
+             = Files.newDirectoryStream(chunksPath.toPath())) {
+      return !dir.iterator().hasNext();
+    }
+  }
+
   /**
    * Parse KeyValueContainerData and verify checksum. Set block related
    * metadata like block commit sequence id, block count, bytes used and

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

@@ -119,6 +119,9 @@ public class OzoneContainer {
   private DatanodeDetails datanodeDetails;
   private StateContext context;
 
+
+  private final ContainerMetrics metrics;
+
   enum InitializingStatus {
     UNINITIALIZED, INITIALIZING, INITIALIZED
   }
@@ -162,7 +165,7 @@ public OzoneContainer(
     metadataScanner = null;
 
     buildContainerSet();
-    final ContainerMetrics metrics = ContainerMetrics.create(conf);
+    metrics = ContainerMetrics.create(conf);
     handlers = Maps.newHashMap();
 
     IncrementalReportSender<Container> icrSender = container -> {
@@ -521,4 +524,8 @@ public MutableVolumeSet getDbVolumeSet() {
   StorageVolumeChecker getVolumeChecker(ConfigurationSource conf) {
     return new StorageVolumeChecker(conf, new Timer());
   }
+
+  public ContainerMetrics getMetrics() {
+    return metrics;
+  }
 }

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueHandler.java

@@ -363,9 +363,7 @@ public void testDeleteContainer() throws IOException {
       Mockito.when(volumeSet.getVolumesList())
           .thenReturn(Collections.singletonList(hddsVolume));
 
-      final int[] interval = new int[1];
-      interval[0] = 2;
-      final ContainerMetrics metrics = new ContainerMetrics(interval);
+      final ContainerMetrics metrics = ContainerMetrics.create(conf);
 
       final AtomicInteger icrReceived = new AtomicInteger(0);