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

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfigKeys.java

@@ -581,6 +581,13 @@ public final class ScmConfigKeys {
 
   public static final String OZONE_AUDIT_LOG_DEBUG_CMD_LIST_SCMAUDIT =
       "ozone.audit.log.debug.cmd.list.scmaudit";
+
+  public static final String OZONE_SCM_CHECK_EMPTY_CONTAINER_ON_DISK_ON_DELETE =
+      "ozone.scm.check.empty.container.delete";
+  public static final Boolean
+      OZONE_SCM_CHECK_EMPTY_CONTAINER_ON_DISK_ON_DELETE_DEFAULT =
+      true;
+
   /**
    * Never constructed.
    */

hadoop-hdds/common/src/main/resources/ozone-default.xml

@@ -3607,4 +3607,12 @@
       without using the optimised DAG based pruning approach
     </description>
   </property>
+  <property>
+    <name>ozone.scm.check.empty.container.delete</name>
+    <value>true</value>
+    <tag>DATANODE</tag>
+    <description>
+      Disallow deletion of containers if on disk there are chunks still present.
+    </description>
+  </property>
 </configuration>

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/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,13 @@ public KeyValueHandler(ConfigurationSource config,
     } catch (Exception e) {
       throw new RuntimeException(e);
     }
+
+    checkIfNoBlockFiles =
+        conf.getBoolean(
+            ScmConfigKeys.OZONE_SCM_CHECK_EMPTY_CONTAINER_ON_DISK_ON_DELETE,
+        ScmConfigKeys.
+            OZONE_SCM_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 +1239,41 @@ 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 (IOException e) {
+      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-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestDeleteContainerHandler.java

@@ -34,8 +34,10 @@
 import org.apache.hadoop.ozone.client.OzoneClient;
 import org.apache.hadoop.ozone.client.OzoneClientFactory;
 import org.apache.hadoop.ozone.client.io.OzoneOutputStream;
+import org.apache.hadoop.ozone.container.common.helpers.ContainerMetrics;
 import org.apache.hadoop.ozone.container.common.impl.ContainerData;
 import org.apache.hadoop.ozone.container.common.interfaces.Container;
+import org.apache.hadoop.ozone.container.keyvalue.KeyValueHandler;
 import org.apache.hadoop.ozone.om.helpers.OmKeyArgs;
 import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
 import org.apache.hadoop.ozone.protocol.commands.CloseContainerCommand;
@@ -47,6 +49,7 @@
 import org.junit.BeforeClass;
 import org.junit.Test;
 
+import java.io.File;
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.UUID;
@@ -107,6 +110,120 @@ public static void shutdown() {
     }
   }
 
+  @Test(timeout = 60000)
+  public void testDeleteNonEmptyContainerDir()
+      throws Exception {
+    // 1. Test if a non force deletion fails if chunks are still present with
+    //    block count set to 0
+    // 2. Test if a force deletion passes even if chunks are still present
+
+    //the easiest way to create an open container is creating a key
+
+    String keyName = UUID.randomUUID().toString();
+
+    // create key
+    createKey(keyName);
+
+    // get containerID of the key
+    ContainerID containerId = getContainerID(keyName);
+
+    ContainerInfo container = cluster.getStorageContainerManager()
+        .getContainerManager().getContainer(containerId);
+
+    Pipeline pipeline = cluster.getStorageContainerManager()
+        .getPipelineManager().getPipeline(container.getPipelineID());
+
+    // We need to close the container because delete container only happens
+    // on closed containers when force flag is set to false.
+
+    HddsDatanodeService hddsDatanodeService =
+        cluster.getHddsDatanodes().get(0);
+
+    Assert.assertFalse(isContainerClosed(hddsDatanodeService,
+        containerId.getId()));
+
+    DatanodeDetails datanodeDetails = hddsDatanodeService.getDatanodeDetails();
+
+    NodeManager nodeManager =
+        cluster.getStorageContainerManager().getScmNodeManager();
+    //send the order to close the container
+    SCMCommand<?> command = new CloseContainerCommand(
+        containerId.getId(), pipeline.getId());
+    command.setTerm(
+        cluster.getStorageContainerManager().getScmContext().getTermOfLeader());
+    nodeManager.addDatanodeCommand(datanodeDetails.getUuid(), command);
+
+    Container containerInternalObj =
+        hddsDatanodeService.
+            getDatanodeStateMachine().
+            getContainer().getContainerSet().getContainer(containerId.getId());
+
+    // Write a file to the container chunks directory indicating that there
+    // might be a discrepancy between block count as recorded in RocksDB and
+    // what is actually on disk.
+    File lingeringBlock =
+        new File(containerInternalObj.
+            getContainerData().getChunksPath() + "/1.block");
+    lingeringBlock.createNewFile();
+    ContainerMetrics metrics =
+        hddsDatanodeService
+            .getDatanodeStateMachine().getContainer().getMetrics();
+    GenericTestUtils.waitFor(() ->
+            isContainerClosed(hddsDatanodeService, containerId.getId()),
+        500, 5 * 1000);
+
+    //double check if it's really closed (waitFor also throws an exception)
+    Assert.assertTrue(isContainerClosed(hddsDatanodeService,
+        containerId.getId()));
+
+    // Check container exists before sending delete container command
+    Assert.assertFalse(isContainerDeleted(hddsDatanodeService,
+        containerId.getId()));
+
+    // Set container blockCount to 0 to mock that it is empty as per RocksDB
+    getContainerfromDN(hddsDatanodeService, containerId.getId())
+        .getContainerData().setBlockCount(0);
+
+    // send delete container to the datanode
+    command = new DeleteContainerCommand(containerId.getId(), false);
+
+    // Send the delete command. It should fail as even though block count
+    // is zero there is a lingering block on disk.
+    command.setTerm(
+        cluster.getStorageContainerManager().getScmContext().getTermOfLeader());
+    nodeManager.addDatanodeCommand(datanodeDetails.getUuid(), command);
+
+    // Deleting a non-empty container should pass on the DN when the force flag
+    // is true
+    // Check the log for the error message when deleting non-empty containers
+    GenericTestUtils.LogCapturer logCapturer =
+        GenericTestUtils.LogCapturer.captureLogs(
+            LoggerFactory.getLogger(KeyValueHandler.class));
+    GenericTestUtils.waitFor(() ->
+            logCapturer.getOutput().contains("container is not empty"),
+        500,
+        5 * 2000);
+    Assert.assertTrue(!isContainerDeleted(hddsDatanodeService,
+        containerId.getId()));
+    Assert.assertEquals(1,
+        metrics.getContainerDeleteFailedNonEmptyDir());
+    // Send the delete command. It should pass with force flag.
+    long beforeForceCount = metrics.getContainerForceDelete();
+    command = new DeleteContainerCommand(containerId.getId(), true);
+
+    command.setTerm(
+        cluster.getStorageContainerManager().getScmContext().getTermOfLeader());
+    nodeManager.addDatanodeCommand(datanodeDetails.getUuid(), command);
+
+    GenericTestUtils.waitFor(() ->
+            isContainerDeleted(hddsDatanodeService, containerId.getId()),
+        500, 5 * 1000);
+    Assert.assertTrue(isContainerDeleted(hddsDatanodeService,
+        containerId.getId()));
+    Assert.assertTrue(beforeForceCount <
+        metrics.getContainerForceDelete());
+  }
+
   @Test(timeout = 60000)
   public void testDeleteContainerRequestHandlerOnClosedContainer()
       throws Exception {
@@ -175,7 +292,11 @@ public void testDeleteContainerRequestHandlerOnClosedContainer()
     GenericTestUtils.waitFor(() -> logCapturer.getOutput().contains("Non" +
         "-force deletion of non-empty container is not allowed"), 500,
         5 * 1000);
-
+    ContainerMetrics metrics =
+        hddsDatanodeService
+            .getDatanodeStateMachine().getContainer().getMetrics();
+    Assert.assertEquals(1,
+        metrics.getContainerDeleteFailedBlockCountNotZero());
     // Set container blockCount to 0 to mock that it is empty
     getContainerfromDN(hddsDatanodeService, containerId.getId())
         .getContainerData().setBlockCount(0);