HDDS-14926. Allow QUASI_CLOSED containers in DiskBalancer with improved debug logs for containers

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

@@ -21,10 +21,18 @@
 
 import jakarta.annotation.Nonnull;
 import java.time.Duration;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Locale;
+import java.util.Set;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.hdds.HddsUtils;
 import org.apache.hadoop.hdds.conf.Config;
 import org.apache.hadoop.hdds.conf.ConfigGroup;
 import org.apache.hadoop.hdds.conf.ConfigTag;
 import org.apache.hadoop.hdds.conf.ConfigType;
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerDataProto.State;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -37,6 +45,12 @@ public final class DiskBalancerConfiguration {
   private static final Logger LOG =
       LoggerFactory.getLogger(DiskBalancerConfiguration.class);
 
+  /**
+   * Default value for {@code hdds.datanode.disk.balancer.container.states}: {@value}
+   * (CLOSED and QUASI_CLOSED containers may be disk-balanced).
+   */
+  public static final String DEFAULT_CONTAINER_STATES = "CLOSED,QUASI_CLOSED";
+
   @Config(key = "hdds.datanode.disk.balancer.info.dir", type = ConfigType.STRING,
       defaultValue = "", tags = {ConfigTag.DISKBALANCER},
       description = "The path where datanode diskBalancer's conf is to be " +
@@ -121,6 +135,19 @@ public final class DiskBalancerConfiguration {
           "Unit could be defined with postfix (ns,ms,s,m,h,d).")
   private long replicaDeletionDelay = Duration.ofMinutes(5).toMillis();
 
+  private static final String HDDS_DATANODE_DISK_BALANCER_CONTAINER_STATES =
+      "hdds.datanode.disk.balancer.container.states";
+
+  @Config(key = HDDS_DATANODE_DISK_BALANCER_CONTAINER_STATES,
+      defaultValue = DEFAULT_CONTAINER_STATES,
+      type = ConfigType.STRING,
+      tags = { DATANODE, ConfigTag.DISKBALANCER },
+      description = "Container lifecycle state names (CLOSED, QUASI_CLOSED) that " +
+          "are eligible to be moved between disks on a datanode. " +
+          "Names must match the enum exactly (uppercase). The default includes CLOSED and QUASI_CLOSED; " +
+          "additional states can be enabled by adding the states here.")
+  private String containerStates = DEFAULT_CONTAINER_STATES;
+
   public DiskBalancerConfiguration(Double threshold,
       Long bandwidthInMB,
       Integer parallelThread,
@@ -262,17 +289,102 @@ public void setParallelThread(int parallelThread) {
     this.parallelThread = parallelThread;
   }
 
+  public String getContainerStates() {
+    return containerStates;
+  }
+
+  /**
+   * Sets the comma-separated container state names and validates them immediately.
+   * <p>
+   * Ozone configuration injection may set the raw container-states string without invoking this
+   * setter; {@link #getMovableContainerStates()} still parses the current field value on read.
+   *
+   * @param containerStates comma-separated {@link State} names (case-sensitive, uppercase);
+   *     whitespace allowed around commas and tokens
+   * @throws IllegalArgumentException if blank, empty after parsing tokens, or unknown state
+   */
+  public void setContainerStates(String containerStates) {
+    parseMovableContainerStates(containerStates);
+    this.containerStates = containerStates;
+  }
+
+  /**
+   * Container lifecycle states eligible for disk balancing, derived from {@link #getContainerStates()}:
+   * same rules as {@link #setContainerStates(String)}.
+   *
+   * @return unmodifiable non-empty set
+   */
+  public Set<State> getMovableContainerStates() {
+    return parseMovableContainerStates(this.containerStates);
+  }
+
+  /**
+   * Parses and validates {@code hdds.datanode.disk.balancer.container.states}.
+   */
+  private static Set<State> parseMovableContainerStates(String raw) {
+    if (StringUtils.isBlank(raw)) {
+      throw new IllegalArgumentException(HDDS_DATANODE_DISK_BALANCER_CONTAINER_STATES + " must not be empty.");
+    }
+    Set<State> states = new HashSet<>();
+    for (String part : raw.split(",")) {
+      String name = part.trim();
+      if (name.isEmpty()) {
+        continue;
+      }
+      State state;
+      try {
+        state = State.valueOf(name);
+      } catch (IllegalArgumentException ex) {
+        if (wouldMatchContainerStateIfUppercased(name)) {
+          throw new IllegalArgumentException(
+              "Invalid container state '" + name + "': use uppercase enum names "
+                  + "(e.g. CLOSED, QUASI_CLOSED). Valid names are: "
+                  + Arrays.toString(State.values()),
+              ex);
+        }
+        throw new IllegalArgumentException(
+            "Invalid container state '" + name + "' in " + HDDS_DATANODE_DISK_BALANCER_CONTAINER_STATES + ". "
+                + "Valid names are: " + Arrays.toString(State.values()),
+            ex);
+      }
+      if (HddsUtils.isOpenToWriteState(state) || state == State.DELETED) {
+        throw new IllegalArgumentException("State " + name + " is not movable.");
+      }
+      states.add(State.valueOf(name));
+    }
+    if (states.isEmpty()) {
+      throw new IllegalArgumentException(
+          HDDS_DATANODE_DISK_BALANCER_CONTAINER_STATES + " must list at least one valid state.");
+    }
+    return Collections.unmodifiableSet(states);
+  }
+
+  private static boolean wouldMatchContainerStateIfUppercased(String name) {
+    String upper = name.toUpperCase(Locale.ROOT);
+    if (upper.equals(name)) {
+      return false;
+    }
+    try {
+      State.valueOf(upper);
+      return true;
+    } catch (IllegalArgumentException e) {
+      return false;
+    }
+  }
+
   @Override
   public String toString() {
     return String.format("Disk Balancer Configuration values:%n" +
             "%-50s %s%n" +
             "%-50s %s%n" +
             "%-50s %s%n" +
             "%-50s %s%n" +
+            "%-50s %s%n" +
             "%-50s %s%n",
             "Key", "Value",
         "Threshold", threshold, "Max disk bandwidth", diskBandwidthInMB,
-        "Parallel Thread", parallelThread, "Stop After Disk Even", stopAfterDiskEven);
+        "Parallel Thread", parallelThread, "Stop After Disk Even", stopAfterDiskEven,
+        "Container states", containerStates);
   }
 
   public HddsProtos.DiskBalancerConfigurationProto.Builder toProtobufBuilder() {
@@ -282,7 +394,8 @@ public HddsProtos.DiskBalancerConfigurationProto.Builder toProtobufBuilder() {
     builder.setThreshold(threshold)
         .setDiskBandwidthInMB(diskBandwidthInMB)
         .setParallelThread(parallelThread)
-        .setStopAfterDiskEven(stopAfterDiskEven);
+        .setStopAfterDiskEven(stopAfterDiskEven)
+        .setContainerStates(containerStates);
     return builder;
   }
 
@@ -309,6 +422,9 @@ public static DiskBalancerConfiguration updateFromProtobuf(
     if (newConfigProto.hasStopAfterDiskEven()) {
       existingConfig.setStopAfterDiskEven(newConfigProto.getStopAfterDiskEven());
     }
+    if (newConfigProto.hasContainerStates()) {
+      existingConfig.setContainerStates(newConfigProto.getContainerStates());
+    }
     return existingConfig;
   }
 }

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

@@ -39,6 +39,7 @@ public class DiskBalancerInfo {
   private long bytesToMove;
   private long balancedBytes;
   private double volumeDataDensity;
+  private String containerStates;
   // Report-only: ideal usage from volume snapshot. NOT persisted.
   private double idealUsage;
   // Report-only: per-volume info. NOT persisted.
@@ -47,29 +48,33 @@ public class DiskBalancerInfo {
   public DiskBalancerInfo(DiskBalancerRunningStatus operationalState, double threshold,
       long bandwidthInMB, int parallelThread, boolean stopAfterDiskEven) {
     this(operationalState, threshold, bandwidthInMB, parallelThread, stopAfterDiskEven,
-        DiskBalancerVersion.DEFAULT_VERSION);
+        DiskBalancerConfiguration.DEFAULT_CONTAINER_STATES, DiskBalancerVersion.DEFAULT_VERSION);
   }
 
   public DiskBalancerInfo(DiskBalancerRunningStatus operationalState, double threshold,
-      long bandwidthInMB, int parallelThread, boolean stopAfterDiskEven, DiskBalancerVersion version) {
+      long bandwidthInMB, int parallelThread, boolean stopAfterDiskEven,
+      String containerStates, DiskBalancerVersion version) {
     this.operationalState = operationalState;
     this.threshold = threshold;
     this.bandwidthInMB = bandwidthInMB;
     this.parallelThread = parallelThread;
     this.stopAfterDiskEven = stopAfterDiskEven;
+    this.containerStates = containerStates;
     this.version = version;
   }
 
   @SuppressWarnings("checkstyle:ParameterNumber")
   public DiskBalancerInfo(DiskBalancerRunningStatus operationalState, double threshold,
       long bandwidthInMB, int parallelThread, boolean stopAfterDiskEven, DiskBalancerVersion version,
-      long successCount, long failureCount, long bytesToMove, long balancedBytes, double volumeDataDensity) {
+      String containerStates, long successCount, long failureCount, long bytesToMove,
+      long balancedBytes, double volumeDataDensity) {
     this.operationalState = operationalState;
     this.threshold = threshold;
     this.bandwidthInMB = bandwidthInMB;
     this.parallelThread = parallelThread;
     this.stopAfterDiskEven = stopAfterDiskEven;
     this.version = version;
+    this.containerStates = containerStates;
     this.successCount = successCount;
     this.failureCount = failureCount;
     this.bytesToMove = bytesToMove;
@@ -88,6 +93,7 @@ public DiskBalancerInfo(boolean shouldRun,
     this.bandwidthInMB = diskBalancerConf.getDiskBandwidthInMB();
     this.parallelThread = diskBalancerConf.getParallelThread();
     this.stopAfterDiskEven = diskBalancerConf.isStopAfterDiskEven();
+    this.containerStates = diskBalancerConf.getContainerStates();
     this.version = DiskBalancerVersion.DEFAULT_VERSION;
   }
 
@@ -104,6 +110,9 @@ public void updateFromConf(DiskBalancerConfiguration diskBalancerConf) {
     if (stopAfterDiskEven != diskBalancerConf.isStopAfterDiskEven()) {
       setStopAfterDiskEven(diskBalancerConf.isStopAfterDiskEven());
     }
+    if (!Objects.equals(containerStates, diskBalancerConf.getContainerStates())) {
+      setContainerStates(diskBalancerConf.getContainerStates());
+    }
   }
 
   /**
@@ -117,9 +126,18 @@ public DiskBalancerConfiguration toConfiguration() {
     config.setDiskBandwidthInMB(this.bandwidthInMB);
     config.setParallelThread(this.parallelThread);
     config.setStopAfterDiskEven(this.stopAfterDiskEven);
+    config.setContainerStates(this.containerStates);
     return config;
   }
 
+  public String getContainerStates() {
+    return containerStates;
+  }
+
+  public void setContainerStates(String containerStates) {
+    this.containerStates = containerStates;
+  }
+
   public DiskBalancerRunningStatus getOperationalState() {
     return operationalState;
   }
@@ -246,12 +264,13 @@ public boolean equals(Object o) {
         bandwidthInMB == that.bandwidthInMB &&
         parallelThread == that.parallelThread &&
         stopAfterDiskEven == that.stopAfterDiskEven &&
-        version == that.version;
+        version == that.version &&
+        Objects.equals(containerStates, that.containerStates);
   }
 
   @Override
   public int hashCode() {
     return Objects.hash(operationalState, threshold, bandwidthInMB, parallelThread, stopAfterDiskEven,
-        version);
+        version, containerStates);
   }
 }

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

@@ -76,7 +76,8 @@ public DatanodeDiskBalancerInfoProto getDiskBalancerInfo(GetDiskBalancerInfoRequ
                 .setThreshold(info.getThreshold())
                 .setDiskBandwidthInMB(info.getBandwidthInMB())
                 .setParallelThread(info.getParallelThread())
-                .setStopAfterDiskEven(info.isStopAfterDiskEven()))
+                .setStopAfterDiskEven(info.isStopAfterDiskEven())
+                .setContainerStates(info.getContainerStates()))
         .setSuccessMoveCount(info.getSuccessCount())
         .setFailureMoveCount(info.getFailureCount())
         .setBytesToMove(info.getBytesToMove())