HDDS-8254. Close containers when volume reaches utilisation threshold

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java

@@ -62,6 +62,11 @@ public final class HddsConfigKeys {
   public static final String HDDS_DATANODE_VOLUME_CHOOSING_POLICY =
       "hdds.datanode.volume.choosing.policy";
 
+  public static final String HDDS_DATANODE_VOLUME_MIN_FREE_SPACE =
+      "hdds.datanode.volume.min.free.space";
+  public static final String HDDS_DATANODE_VOLUME_MIN_FREE_SPACE_DEFAULT =
+      "5GB";
+
   public static final String HDDS_DB_PROFILE = "hdds.db.profile";
 
   // Once a container usage crosses this threshold, it is eligible for

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

@@ -204,6 +204,17 @@
       This volume choosing policy selects volumes in a round-robin order.
     </description>
   </property>
+  <property>
+    <name>hdds.datanode.volume.min.free.space</name>
+    <value>5GB</value>
+    <tag>OZONE, CONTAINER, STORAGE, MANAGEMENT</tag>
+    <description>
+      This determines the free space to be used for closing containers
+      When the difference between volume capacity and used reaches this number,
+      containers that reside on this volume will be closed and no new containers
+      would be allocated on this volume.
+    </description>
+  </property>
   <property>
     <name>dfs.container.ratis.enabled</name>
     <value>false</value>

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/HddsDispatcher.java

@@ -24,6 +24,7 @@
 import org.apache.hadoop.hdds.HddsConfigKeys;
 import org.apache.hadoop.hdds.HddsUtils;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.conf.StorageUnit;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandRequestProto;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandResponseProto;
@@ -556,11 +557,18 @@ private boolean isContainerFull(Container container) {
     boolean isOpen = Optional.ofNullable(container)
         .map(cont -> cont.getContainerState() == ContainerDataProto.State.OPEN)
         .orElse(Boolean.FALSE);
+    long volumeFreeSpaceToSpare = (long) conf.getStorageSize(
+        HddsConfigKeys.HDDS_DATANODE_VOLUME_MIN_FREE_SPACE,
+        HddsConfigKeys.HDDS_DATANODE_VOLUME_MIN_FREE_SPACE_DEFAULT,
+        StorageUnit.BYTES);
     if (isOpen) {
       ContainerData containerData = container.getContainerData();
       double containerUsedPercentage =
           1.0f * containerData.getBytesUsed() / containerData.getMaxSize();
-      return containerUsedPercentage >= containerCloseThreshold;
+      float volumeUsed = containerData.getVolume().getUsedSpace();
+      float volumeCapacity = containerData.getVolume().getCapacity();
+      return (containerUsedPercentage >= containerCloseThreshold) ||
+          (volumeCapacity - volumeUsed <= volumeFreeSpaceToSpare);
     } else {
       return false;
     }

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

@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.ozone.container.common.interfaces;
 
+import com.google.common.annotations.VisibleForTesting;
 import org.apache.hadoop.hdds.annotation.InterfaceAudience;
 import org.apache.hadoop.ozone.container.common.volume.HddsVolume;
 
@@ -38,9 +39,17 @@ public interface VolumeChoosingPolicy {
    * @param volumes - a list of available volumes.
    * @param maxContainerSize - the maximum size of the container for which a
    *                         volume is sought.
+   * @param volumeFreeSpace - Free space to spare on the volume ie when
+   *                        (capacity-used) reaches volumeFreeSpace, the volume
+   *                        should be ineligible for container allocation.
+   *
    * @return the chosen volume.
    * @throws IOException when disks are unavailable or are full.
    */
+  HddsVolume chooseVolume(List<HddsVolume> volumes, long maxContainerSize,
+      long volumeFreeSpace) throws IOException;
+
+  @VisibleForTesting
   HddsVolume chooseVolume(List<HddsVolume> volumes, long maxContainerSize)
       throws IOException;
 }

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

@@ -32,16 +32,23 @@ class AvailableSpaceFilter implements Predicate<HddsVolume> {
       new HashMap<>();
   private long mostAvailableSpace = Long.MIN_VALUE;
 
-  AvailableSpaceFilter(long requiredSpace) {
+  private long volumeFreeSpace;
+
+  AvailableSpaceFilter(long requiredSpace, long volumeFreeSpace) {
     this.requiredSpace = requiredSpace;
+    this.volumeFreeSpace = volumeFreeSpace;
   }
 
   @Override
   public boolean test(HddsVolume vol) {
+    float volumeUsed = vol.getUsedSpace();
+    float volumeCapacity = vol.getCapacity();
     long free = vol.getAvailable();
     long committed = vol.getCommittedBytes();
     long available = free - committed;
-    boolean hasEnoughSpace = available > requiredSpace;
+    boolean hasEnoughSpace =
+        (available > requiredSpace) &&
+            (volumeCapacity - volumeUsed > volumeFreeSpace);
 
     mostAvailableSpace = Math.max(available, mostAvailableSpace);
 

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

@@ -52,13 +52,21 @@ public class CapacityVolumeChoosingPolicy implements VolumeChoosingPolicy {
   @Override
   public HddsVolume chooseVolume(List<HddsVolume> volumes,
       long maxContainerSize) throws IOException {
+    return chooseVolume(volumes, maxContainerSize, 0L);
+  }
+
+  @Override
+  public HddsVolume chooseVolume(List<HddsVolume> volumes,
+      long maxContainerSize, long volumeFreeSpace)
+      throws IOException {
 
     // No volumes available to choose from
     if (volumes.isEmpty()) {
       throw new DiskOutOfSpaceException("No more available volumes");
     }
 
-    AvailableSpaceFilter filter = new AvailableSpaceFilter(maxContainerSize);
+    AvailableSpaceFilter filter = new AvailableSpaceFilter(maxContainerSize,
+        volumeFreeSpace);
 
     List<HddsVolume> volumesWithEnoughSpace = volumes.stream()
         .filter(filter)

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

@@ -42,16 +42,25 @@ public class RoundRobinVolumeChoosingPolicy implements VolumeChoosingPolicy {
   // Stores the index of the next volume to be returned.
   private AtomicInteger nextVolumeIndex = new AtomicInteger(0);
 
+
   @Override
   public HddsVolume chooseVolume(List<HddsVolume> volumes,
       long maxContainerSize) throws IOException {
+    return chooseVolume(volumes, maxContainerSize, 0L);
+  }
+
+  @Override
+  public HddsVolume chooseVolume(List<HddsVolume> volumes,
+      long maxContainerSize, long volumeFreeSpace)
+      throws IOException {
 
     // No volumes available to choose from
     if (volumes.size() < 1) {
       throw new DiskOutOfSpaceException("No more available volumes");
     }
 
-    AvailableSpaceFilter filter = new AvailableSpaceFilter(maxContainerSize);
+    AvailableSpaceFilter filter =
+        new AvailableSpaceFilter(maxContainerSize, volumeFreeSpace);
 
     // since volumes could've been removed because of the failure
     // make sure we are not out of bounds

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

@@ -35,8 +35,10 @@
 
 import org.apache.hadoop.fs.FileAlreadyExistsException;
 import org.apache.hadoop.fs.FileUtil;
+import org.apache.hadoop.hdds.HddsConfigKeys;
 import org.apache.hadoop.hdds.HddsUtils;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.conf.StorageUnit;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerDataProto;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerType;
@@ -136,11 +138,15 @@ public void create(VolumeSet volumeSet, VolumeChoosingPolicy
     File containerMetaDataPath = null;
     //acquiring volumeset read lock
     long maxSize = containerData.getMaxSize();
+    long volumeFreeSpaceToSpare = (long) config.getStorageSize(
+        HddsConfigKeys.HDDS_DATANODE_VOLUME_MIN_FREE_SPACE,
+        HddsConfigKeys.HDDS_DATANODE_VOLUME_MIN_FREE_SPACE_DEFAULT,
+        StorageUnit.BYTES);
     volumeSet.readLock();
     try {
       HddsVolume containerVolume = volumeChoosingPolicy.chooseVolume(
           StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()),
-          maxSize);
+          maxSize, volumeFreeSpaceToSpare);
       String hddsVolumeDir = containerVolume.getHddsRootDir().toString();
       // Set volume before getContainerDBFile(), because we may need the
       // volume to deduce the db file.

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

@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.ozone.container.replication;
 
+import org.apache.hadoop.hdds.HddsConfigKeys;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
 import org.apache.hadoop.hdds.conf.StorageUnit;
 import org.apache.hadoop.hdds.scm.ScmConfigKeys;
@@ -57,6 +58,7 @@ public class ContainerImporter {
   private final MutableVolumeSet volumeSet;
   private final VolumeChoosingPolicy volumeChoosingPolicy;
   private final long containerSize;
+  private final long volumeFreeSpace;
 
   public ContainerImporter(ConfigurationSource conf, ContainerSet containerSet,
       ContainerController controller,
@@ -74,6 +76,10 @@ public ContainerImporter(ConfigurationSource conf, ContainerSet containerSet,
     containerSize = (long) conf.getStorageSize(
         ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE,
         ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE_DEFAULT, StorageUnit.BYTES);
+    volumeFreeSpace = (long) conf.getStorageSize(
+        HddsConfigKeys.HDDS_DATANODE_VOLUME_MIN_FREE_SPACE,
+        HddsConfigKeys.HDDS_DATANODE_VOLUME_MIN_FREE_SPACE_DEFAULT,
+        StorageUnit.BYTES);
   }
 
   public void importContainer(long containerID, Path tarFilePath,
@@ -116,7 +122,7 @@ HddsVolume chooseNextVolume() throws IOException {
     // Choose volume that can hold both container in tmp and dest directory
     return volumeChoosingPolicy.chooseVolume(
         StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()),
-        containerSize * 2);
+        containerSize * 2, volumeFreeSpace);
   }
 
   public static Path getUntarDirectory(HddsVolume hddsVolume)

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

@@ -114,7 +114,8 @@ public void init() throws IOException {
     volumeSet = mock(MutableVolumeSet.class);
 
     volumeChoosingPolicy = mock(RoundRobinVolumeChoosingPolicy.class);
-    Mockito.when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong()))
+    Mockito.when(
+            volumeChoosingPolicy.chooseVolume(anyList(), anyLong(), anyLong()))
         .thenReturn(hddsVolume);
   }
 

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestHddsDispatcher.java

@@ -22,8 +22,12 @@
 import org.apache.commons.codec.digest.DigestUtils;
 import org.apache.commons.io.FileUtils;
 import org.apache.hadoop.conf.StorageUnit;
+import org.apache.hadoop.hdds.HddsConfigKeys;
 import org.apache.hadoop.hdds.client.BlockID;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.fs.MockSpaceUsageCheckFactory;
+import org.apache.hadoop.hdds.fs.SpaceUsageCheckFactory;
+import org.apache.hadoop.hdds.fs.SpaceUsageSource;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
 import org.apache.hadoop.hdds.protocol.datanode.proto
@@ -36,6 +40,7 @@
     .WriteChunkRequestProto;
 import org.apache.hadoop.hdds.protocol.proto
     .StorageContainerDatanodeProtocolProtos.ContainerAction;
+import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
 import org.apache.hadoop.ozone.OzoneConfigKeys;
 import org.apache.hadoop.ozone.common.Checksum;
 import org.apache.hadoop.ozone.common.utils.BufferUtils;
@@ -50,12 +55,14 @@
 import org.apache.hadoop.ozone.container.common.volume.RoundRobinVolumeChoosingPolicy;
 import org.apache.hadoop.ozone.container.common.volume.StorageVolume;
 import org.apache.hadoop.ozone.container.common.volume.VolumeSet;
+import org.apache.hadoop.ozone.container.common.volume.HddsVolume;
 import org.apache.hadoop.ozone.container.common.volume.MutableVolumeSet;
 import org.apache.hadoop.ozone.container.keyvalue.ContainerLayoutTestInfo;
 import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainer;
 import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainerData;
 import org.apache.ozone.test.GenericTestUtils;
 
+import org.apache.ozone.test.LambdaTestUtils;
 import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
 import org.junit.Assert;
 import org.junit.Test;
@@ -68,8 +75,13 @@
 import java.util.Collections;
 import java.util.Map;
 import java.util.UUID;
+import java.util.concurrent.atomic.AtomicLong;
+
+import java.time.Duration;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.hadoop.hdds.fs.MockSpaceUsagePersistence.inMemory;
+import static org.apache.hadoop.hdds.fs.MockSpaceUsageSource.fixed;
 import static org.apache.hadoop.hdds.scm.ScmConfigKeys.HDDS_DATANODE_DIR_KEY;
 import static org.apache.hadoop.hdds.scm.protocolPB.ContainerCommandResponseBuilders.getContainerCommandResponse;
 import static org.junit.Assert.assertTrue;
@@ -158,6 +170,86 @@ public void testContainerCloseActionWhenFull() throws IOException {
 
   }
 
+  @Test
+  public void testContainerCloseActionWhenVolumeFull() throws Exception {
+    String testDir = GenericTestUtils.getTempPath(
+        TestHddsDispatcher.class.getSimpleName());
+    OzoneConfiguration conf = new OzoneConfiguration();
+    conf.setStorageSize(HddsConfigKeys.HDDS_DATANODE_VOLUME_MIN_FREE_SPACE,
+        100.0, StorageUnit.BYTES);
+    DatanodeDetails dd = randomDatanodeDetails();
+
+    HddsVolume.Builder volumeBuilder =
+        new HddsVolume.Builder(testDir).datanodeUuid(dd.getUuidString())
+            .conf(conf).usageCheckFactory(MockSpaceUsageCheckFactory.NONE);
+    // state of cluster : capacity-used (140) > 100  ,datanode volume
+    // utilisation threshold not yet reached. container creates are successful.
+    SpaceUsageSource spaceUsage = fixed(500, 140, 360);
+
+    SpaceUsageCheckFactory factory = MockSpaceUsageCheckFactory.of(
+        spaceUsage, Duration.ZERO, inMemory(new AtomicLong(0)));
+    volumeBuilder.usageCheckFactory(factory);
+    MutableVolumeSet volumeSet = Mockito.mock(MutableVolumeSet.class);
+    Mockito.when(volumeSet.getVolumesList())
+        .thenReturn(Collections.singletonList(volumeBuilder.build()));
+    try {
+      UUID scmId = UUID.randomUUID();
+      ContainerSet containerSet = new ContainerSet(1000);
+
+      DatanodeStateMachine stateMachine = Mockito.mock(
+          DatanodeStateMachine.class);
+      StateContext context = Mockito.mock(StateContext.class);
+      Mockito.when(stateMachine.getDatanodeDetails()).thenReturn(dd);
+      Mockito.when(context.getParent()).thenReturn(stateMachine);
+      // create a 50 byte container
+      KeyValueContainerData containerData = new KeyValueContainerData(1L,
+          layout,
+           50, UUID.randomUUID().toString(),
+          dd.getUuidString());
+      Container container = new KeyValueContainer(containerData, conf);
+      container.create(volumeSet, new RoundRobinVolumeChoosingPolicy(),
+          scmId.toString());
+      containerSet.addContainer(container);
+      ContainerMetrics metrics = ContainerMetrics.create(conf);
+      Map<ContainerType, Handler> handlers = Maps.newHashMap();
+      for (ContainerType containerType : ContainerType.values()) {
+        handlers.put(containerType,
+            Handler.getHandlerForContainerType(containerType, conf,
+                context.getParent().getDatanodeDetails().getUuidString(),
+                containerSet, volumeSet, metrics, NO_OP_ICR_SENDER));
+      }
+      HddsDispatcher hddsDispatcher = new HddsDispatcher(
+          conf, containerSet, volumeSet, handlers, context, metrics, null);
+      hddsDispatcher.setClusterId(scmId.toString());
+      containerData.getVolume().getVolumeInfo()
+          .ifPresent(volumeInfo -> volumeInfo.incrementUsedSpace(50));
+      ContainerCommandResponseProto response = hddsDispatcher
+          .dispatch(getWriteChunkRequest(dd.getUuidString(), 1L, 1L), null);
+      Assert.assertEquals(ContainerProtos.Result.SUCCESS,
+          response.getResult());
+      verify(context, times(1))
+          .addContainerActionIfAbsent(Mockito.any(ContainerAction.class));
+
+      // try creating another container now as the volume used has crossed
+      // threshold
+
+      KeyValueContainerData containerData2 = new KeyValueContainerData(1L,
+          layout,
+          50, UUID.randomUUID().toString(),
+          dd.getUuidString());
+      Container container2 = new KeyValueContainer(containerData2, conf);
+      LambdaTestUtils.intercept(StorageContainerException.class,
+          "Container creation failed, due to disk out of space",
+          () -> container2.create(volumeSet,
+              new RoundRobinVolumeChoosingPolicy(), scmId.toString()));
+
+    } finally {
+      volumeSet.shutdown();
+      ContainerMetrics.remove();
+      FileUtils.deleteDirectory(new File(testDir));
+    }
+  }
+
   @Test
   public void testCreateContainerWithWriteChunk() throws IOException {
     String testDir =