HDDS-7058. EC: ReplicationManager - Implement ratis container replication check handler

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ContainerHealthResult.java

@@ -106,6 +106,9 @@ public static class UnderReplicatedHealthResult
     private final int remainingRedundancy;
     private final boolean dueToDecommission;
     private final boolean sufficientlyReplicatedAfterPending;
+    private boolean dueToMisReplication = false;
+    private boolean isMisReplicated = false;
+    private boolean isMisReplicatedAfterPending = false;
     private final boolean unrecoverable;
     private int requeueCount = 0;
 
@@ -119,6 +122,44 @@ public UnderReplicatedHealthResult(ContainerInfo containerInfo,
       this.unrecoverable = unrecoverable;
     }
 
+    /**
+     * Pass true to indicate the container is mis-replicated - ie it does not
+     * meet the placement policy.
+     * @param isMisRep True if the container is mis-replicated, false if not.
+     * @return this object to allow calls to be chained
+     */
+    public UnderReplicatedHealthResult
+        setMisReplicated(boolean isMisRep) {
+      this.isMisReplicated = isMisRep;
+      return this;
+    }
+
+    /**
+     * Pass true to indicate the container is mis-replicated after considering
+     * pending replicas scheduled for create or delete.
+     * @param isMisRep True if the container is mis-replicated considering
+     *                 pending replicas, or false if not.
+     * @return this object to allow calls to be chained
+     */
+    public UnderReplicatedHealthResult
+        setMisReplicatedAfterPending(boolean isMisRep) {
+      this.isMisReplicatedAfterPending = isMisRep;
+      return this;
+    }
+
+    /**
+     * If the container is ONLY under replicated due to mis-replication, pass
+     * true, otherwise pass false.
+     * @param dueToMisRep Pass true if the container has enough replicas but
+     *                    does not meet the placement policy.
+     * @return
+     */
+    public UnderReplicatedHealthResult
+        setDueToMisReplication(boolean dueToMisRep) {
+      this.dueToMisReplication = dueToMisRep;
+      return this;
+    }
+
     /**
      * How many more replicas can be lost before the container is
      * unreadable. For containers which are under-replicated due to decommission
@@ -187,6 +228,34 @@ public boolean isSufficientlyReplicatedAfterPending() {
       return sufficientlyReplicatedAfterPending;
     }
 
+    /**
+     * Returns true if the container is mis-replicated, ignoring any pending
+     * replicas scheduled to be created.
+     * @return True if mis-replicated, ignoring pending
+     */
+    public boolean isMisReplicated() {
+      return isMisReplicated;
+    }
+
+    /**
+     * Returns true if the container is mis-replicated after taking account of
+     * pending replicas, which are schedule to be created.
+     * @return true is mis-replicated after pending.
+     */
+    public boolean isMisReplicatedAfterPending() {
+      return isMisReplicatedAfterPending;
+    }
+
+    /**
+     * Returns true if the under replication is only due to mis-replication.
+     * In other words, the container has enough replicas, but they do not meet
+     * the placement policy.
+     * @return true if the under-replication is only due to mis-replication
+     */
+    public boolean isDueToMisReplication() {
+      return dueToMisReplication;
+    }
+
     /**
      * Indicates whether a container has enough replicas to be read. For Ratis
      * at least one replia must be available. For EC, at least dataNum replicas

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/LegacyReplicationManager.java

@@ -34,7 +34,6 @@
 import org.apache.hadoop.hdds.scm.PlacementPolicy;
 import org.apache.hadoop.hdds.scm.ScmConfigKeys;
 import org.apache.hadoop.hdds.scm.container.ContainerID;
-import org.apache.hadoop.hdds.scm.container.RatisContainerReplicaCount;
 import org.apache.hadoop.hdds.scm.container.ContainerInfo;
 import org.apache.hadoop.hdds.scm.container.ContainerManager;
 import org.apache.hadoop.hdds.scm.container.ContainerNotFoundException;

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/RatisContainerReplicaCount.java → hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/RatisContainerReplicaCount.java

@@ -15,10 +15,11 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.hadoop.hdds.scm.container;
+package org.apache.hadoop.hdds.scm.container.replication;
 
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
-import org.apache.hadoop.hdds.scm.container.replication.ContainerReplicaCount;
+import org.apache.hadoop.hdds.scm.container.ContainerInfo;
+import org.apache.hadoop.hdds.scm.container.ContainerReplica;
 
 import java.util.Set;
 
@@ -243,11 +244,27 @@ private int missingReplicas() {
    */
   @Override
   public boolean isSufficientlyReplicated() {
-    return missingReplicas() + inFlightDel <= 0;
+    return isSufficientlyReplicated(false);
   }
 
   /**
-   * Return true is the container is over replicated. Decommission and
+   * Return true if the container is sufficiently replicated. Decommissioning
+   * and Decommissioned containers are ignored in this check, assuming they will
+   * eventually be removed from the cluster.
+   * This check ignores inflight additions, if includePendingAdd is false,
+   * otherwise it will assume they complete ok.
+   *
+   * @return True if the container is sufficiently replicated and False
+   *         otherwise.
+   */
+  public boolean isSufficientlyReplicated(boolean includePendingAdd) {
+    // Positive for under-rep, negative for over-rep
+    int delta = redundancyDelta(true, includePendingAdd);
+    return delta <= 0;
+  }
+
+  /**
+   * Return true if the container is over replicated. Decommission and
    * maintenance containers are ignored for this check.
    * The check ignores inflight additions, as they may fail, but it does
    * consider inflight deletes, as they would reduce the over replication when
@@ -257,7 +274,67 @@ public boolean isSufficientlyReplicated() {
    */
   @Override
   public boolean isOverReplicated() {
-    return missingReplicas() + inFlightDel < 0;
+    return isOverReplicated(true);
+  }
+
+  /**
+   * Return true if the container is over replicated. Decommission and
+   * maintenance containers are ignored for this check.
+   * The check ignores inflight additions, as they may fail, but it does
+   * consider inflight deletes if includePendingDelete is true.
+   *
+   * @return True if the container is over replicated, false otherwise.
+   */
+  public boolean isOverReplicated(boolean includePendingDelete) {
+    return getExcessRedundancy(includePendingDelete) > 0;
+  }
+
+  /**
+   * @return Return Excess Redundancy replica nums.
+   */
+  public int getExcessRedundancy(boolean includePendingDelete) {
+    int excessRedundancy = redundancyDelta(includePendingDelete, false);
+    if (excessRedundancy >= 0) {
+      // either perfectly replicated or under replicated
+      return 0;
+    }
+    return -excessRedundancy;
+  }
+
+  /**
+   * Return the delta from the expected number of replicas, optionally
+   * considering inflight add and deletes.
+   * @param includePendingDelete
+   * @param includePendingAdd
+   * @return zero if perfectly replicated, a negative value for over replication
+   *         and a positive value for under replication. The magnitude of the
+   *         return value indicates how many replias the container is over or
+   *         under replicated by.
+   */
+  private int redundancyDelta(boolean includePendingDelete,
+      boolean includePendingAdd) {
+    int excessRedundancy = missingReplicas();
+    if (includePendingDelete) {
+      excessRedundancy += inFlightDel;
+    }
+    if (includePendingAdd) {
+      excessRedundancy -= inFlightAdd;
+    }
+    return excessRedundancy;
+  }
+
+  /**
+   * How many more replicas can be lost before the container is
+   * unreadable, assuming any infligh deletes will complete. For containers
+   * which are under-replicated due to decommission
+   * or maintenance only, the remaining redundancy will include those
+   * decommissioning or maintenance replicas, as they are technically still
+   * available until the datanode processes are stopped.
+   * @return Count of remaining redundant replicas.
+   */
+  public int getRemainingRedundancy() {
+    return Math.max(0,
+        healthyCount + decommissionCount + maintenanceCount - inFlightDel - 1);
   }
 
   /**

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java

@@ -41,6 +41,7 @@
 import org.apache.hadoop.hdds.scm.container.replication.health.HealthCheck;
 import org.apache.hadoop.hdds.scm.container.replication.health.OpenContainerHandler;
 import org.apache.hadoop.hdds.scm.container.replication.health.QuasiClosedContainerHandler;
+import org.apache.hadoop.hdds.scm.container.replication.health.RatisReplicationCheckHandler;
 import org.apache.hadoop.hdds.scm.events.SCMEvents;
 import org.apache.hadoop.hdds.scm.ha.SCMContext;
 import org.apache.hadoop.hdds.scm.ha.SCMService;
@@ -72,6 +73,7 @@
 import static org.apache.hadoop.hdds.conf.ConfigTag.OZONE;
 import static org.apache.hadoop.hdds.conf.ConfigTag.SCM;
 import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType.EC;
+import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType.RATIS;
 
 /**
  * Replication Manager (RM) is the one which is responsible for making sure
@@ -142,12 +144,14 @@ public class ReplicationManager implements SCMService {
   private final Clock clock;
   private final ContainerReplicaPendingOps containerReplicaPendingOps;
   private final ECReplicationCheckHandler ecReplicationCheckHandler;
+  private final RatisReplicationCheckHandler ratisReplicationCheckHandler;
   private final EventPublisher eventPublisher;
   private final ReentrantLock lock = new ReentrantLock();
   private ReplicationQueue replicationQueue;
   private final ECUnderReplicationHandler ecUnderReplicationHandler;
   private final ECOverReplicationHandler ecOverReplicationHandler;
   private final int maintenanceRedundancy;
+  private final int ratisMaintenanceMinReplicas;
   private Thread underReplicatedProcessorThread;
   private Thread overReplicatedProcessorThread;
   private final UnderReplicatedProcessor underReplicatedProcessor;
@@ -188,9 +192,13 @@ public ReplicationManager(final ConfigurationSource conf,
     this.containerReplicaPendingOps = replicaPendingOps;
     this.legacyReplicationManager = legacyReplicationManager;
     this.ecReplicationCheckHandler = new ECReplicationCheckHandler();
+    this.ratisReplicationCheckHandler =
+        new RatisReplicationCheckHandler(containerPlacement);
     this.nodeManager = nodeManager;
     this.replicationQueue = new ReplicationQueue();
     this.maintenanceRedundancy = rmConf.maintenanceRemainingRedundancy;
+    this.ratisMaintenanceMinReplicas = rmConf.getMaintenanceReplicaMinimum();
+
     ecUnderReplicationHandler = new ECUnderReplicationHandler(
         ecReplicationCheckHandler, containerPlacement, conf, nodeManager);
     ecOverReplicationHandler =
@@ -210,7 +218,8 @@ public ReplicationManager(final ConfigurationSource conf,
         .addNext(new ClosingContainerHandler(this))
         .addNext(new QuasiClosedContainerHandler(this))
         .addNext(new ClosedWithMismatchedReplicasHandler(this))
-        .addNext(ecReplicationCheckHandler);
+        .addNext(ecReplicationCheckHandler)
+        .addNext(ratisReplicationCheckHandler);
     start();
   }
 
@@ -433,10 +442,16 @@ protected void processContainer(ContainerInfo containerInfo,
     List<ContainerReplicaOp> pendingOps =
         containerReplicaPendingOps.getPendingOps(containerID);
 
+    // There is a different config for EC and Ratis maintenance
+    // minimum replicas, so we must pass through the correct one.
+    int maintRedundancy = maintenanceRedundancy;
+    if (containerInfo.getReplicationType() == RATIS) {
+      maintRedundancy = ratisMaintenanceMinReplicas;
+    }
     ContainerCheckRequest checkRequest = new ContainerCheckRequest.Builder()
         .setContainerInfo(containerInfo)
         .setContainerReplicas(replicas)
-        .setMaintenanceRedundancy(maintenanceRedundancy)
+        .setMaintenanceRedundancy(maintRedundancy)
         .setReport(report)
         .setPendingOps(pendingOps)
         .setReplicationQueue(repQueue)