HDDS-6697. EC: ReplicationManager - create class to detect EC container health issues

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

@@ -63,6 +63,7 @@ public class ECContainerReplicaCount {
   private final ContainerInfo containerInfo;
   private final ECReplicationConfig repConfig;
   private final List<Integer> pendingAdd;
+  private final List<Integer> pendingDelete;
   private final int remainingMaintenanceRedundancy;
   private final Map<Integer, Integer> healthyIndexes = new HashMap<>();
   private final Map<Integer, Integer> decommissionIndexes = new HashMap<>();
@@ -74,6 +75,7 @@ public ECContainerReplicaCount(ContainerInfo containerInfo,
     this.containerInfo = containerInfo;
     this.repConfig = (ECReplicationConfig)containerInfo.getReplicationConfig();
     this.pendingAdd = indexesPendingAdd;
+    this.pendingDelete = indexesPendingDelete;
     this.remainingMaintenanceRedundancy
         = Math.min(repConfig.getParity(), remainingMaintenanceRedundancy);
 
@@ -152,15 +154,21 @@ public boolean unRecoverable() {
   /**
    * Returns an unsorted list of indexes which need additional copies to
    * ensure the container is sufficiently replicated. These missing indexes will
-   * not be on maintenance nodes, although they may be on decommissioning nodes.
-   * Replicas pending delete are assumed to be removed and any pending add
-   * are assume to be created and omitted them from the returned list. This list
-   * can be used to determine which replicas must be recovered in a group,
-   * assuming the inflight replicas pending add complete successfully.
-   * @return List of missing indexes
+   * not be on maintenance nodes, or decommission nodes.
+   * Replicas pending delete are assumed to be removed.
+   * If includePendingAdd is true, any replicas pending add
+   * are assume to be created and omitted them from the returned list. If it is
+   * true, we assume the pendingAdd will complete, giving a view of the
+   * potential future state of the container.
+   * This list can be used to determine which replicas must be recovered via an
+   * EC reconstuction, rather than making copies of maintenance / decommission
+   * replicas
+   * @param includePendingAdd  If true, treat pending add containers as if they
+   *                           have completed successfully.
+   * @return List of missing indexes which have no online copy.
    */
-  public List<Integer> missingNonMaintenanceIndexes() {
-    if (isSufficientlyReplicated()) {
+  public List<Integer> unavailableIndexes(boolean includePendingAdd) {
+    if (isSufficientlyReplicated(false)) {
       return Collections.emptyList();
     }
     Set<Integer> missing = new HashSet<>();
@@ -171,14 +179,20 @@ public List<Integer> missingNonMaintenanceIndexes() {
     }
     // Now we have a list of missing. Remove any pending add as they should
     // eventually recover.
-    for (Integer i : pendingAdd) {
-      missing.remove(i);
+    if (includePendingAdd) {
+      for (Integer i : pendingAdd) {
+        missing.remove(i);
+      }
     }
     // Remove any maintenance copies, as they are still available. What remains
     // is the set of indexes we have no copy of, and hence must get re-created
     for (Integer i : maintenanceIndexes.keySet()) {
       missing.remove(i);
     }
+    // Remove any decommission copies, as they are still available
+    for (Integer i : decommissionIndexes.keySet()) {
+      missing.remove(i);
+    }
     return missing.stream().collect(Collectors.toList());
   }
 
@@ -214,16 +228,22 @@ public int additionalMaintenanceCopiesNeeded() {
 
   /**
    * If any index has more than one copy that is not in maintenance or
-   * decommission, then the container is over replicated. We consider inflight
-   * deletes, assuming they will be removed. Inflight adds are ignored until
-   * they are actually created.
+   * decommission, then the container is over replicated. If the
+   * includePendingDeletes flag is false we ignore replicas pending delete.
+   * If it is true, we assume inflight deletes have been removed, giving
+   * a view of the future state of the container if they complete successfully.
+   * Pending add are always ignored as they may fail to create.
    * Note it is possible for a container to be both over and under replicated
    * as it could have multiple copies of 1 index, but zero copies of another
    * index.
+   * @param includePendingDelete If true, treat replicas pending delete as if
+   *                             they have deleted successfully.
    * @return True if overReplicated, false otherwise.
    */
-  public boolean isOverReplicated() {
-    for (Integer count : healthyIndexes.values()) {
+  public boolean isOverReplicated(boolean includePendingDelete) {
+    final Map<Integer, Integer> availableIndexes
+        = getHealthyWithDelete(includePendingDelete);
+    for (Integer count : availableIndexes.values()) {
       if (count > 1) {
         return true;
       }
@@ -237,44 +257,71 @@ public boolean isOverReplicated() {
    * as if we have excess including maintenance, it may be due to replication
    * which was needed to ensure sufficient redundancy for maintenance.
    * Pending adds are ignored as they may fail to complete.
+   * If the includePendingDeletes flag is false we ignore replicas pending
+   * delete. If it is true, we assume inflight deletes have been removed, giving
+   * a view of the future state of the container if they complete successfully.
    * Pending deletes are assumed to complete and any indexes returned from here
    * will have the pending deletes already removed.
+   * @param includePendingDelete If true, treat replicas pending delete as if
+   *    *                        they have deleted successfully.
    * @return List of indexes which are over-replicated.
    */
-  public List<Integer> overReplicatedIndexes() {
+  public List<Integer> overReplicatedIndexes(boolean includePendingDelete) {
+    final Map<Integer, Integer> availableIndexes
+        = getHealthyWithDelete(includePendingDelete);
     List<Integer> indexes = new ArrayList<>();
-    for (Map.Entry<Integer, Integer> entry : healthyIndexes.entrySet()) {
+    for (Map.Entry<Integer, Integer> entry : availableIndexes.entrySet()) {
       if (entry.getValue() > 1) {
         indexes.add(entry.getKey());
       }
     }
     return indexes;
   }
 
+  private Map<Integer, Integer> getHealthyWithDelete(boolean includeDelete) {
+    final Map<Integer, Integer> availableIndexes;
+    if (includeDelete) {
+      // Deletes are already removed from the healthy list so just use the
+      // healthy list
+      availableIndexes = Collections.unmodifiableMap(healthyIndexes);
+    } else {
+      availableIndexes = new HashMap<>(healthyIndexes);
+      pendingDelete.forEach(k -> availableIndexes.merge(k, 1, Integer::sum));
+    }
+    return availableIndexes;
+  }
+
   /**
    * The container is sufficiently replicated if the healthy indexes minus any
    * pending deletes give a complete set of container indexes. If not, we must
    * also check the maintenance indexes - the container is still sufficiently
    * replicated if the complete set is made up of healthy + maintenance and
    * there is still sufficient maintenance redundancy.
+   * If the includePendingAdd flag is set to true, this method treats replicas
+   * pending add as if they have completed and hence shows the potential future
+   * state of the container assuming they all complete.
+   * @param includePendingAdd If true, treat pending add containers as if they
+   *                          have completed successfully.
    * @return True if sufficiently replicated, false otherwise.
    */
-  public boolean isSufficientlyReplicated() {
-    if (hasFullSetOfIndexes(healthyIndexes)) {
-      return true;
+  public boolean isSufficientlyReplicated(boolean includePendingAdd) {
+    final Map<Integer, Integer> onlineIndexes;
+    if (includePendingAdd) {
+      onlineIndexes = new HashMap<>(healthyIndexes);
+      pendingAdd.forEach(k -> onlineIndexes.merge(k, 1, Integer::sum));
+    } else {
+      onlineIndexes = Collections.unmodifiableMap(healthyIndexes);
     }
-    // If we don't have a full healthy set, we could have some maintenance
-    // replicas that make up the full set.
-    // For maintenance, we must have at least dataNum + maintenance redundancy
-    // available.
-    if (healthyIndexes.size() <
-        repConfig.getData() + remainingMaintenanceRedundancy) {
-      return false;
+
+    if (hasFullSetOfIndexes(onlineIndexes)) {
+      return true;
     }
-    // Finally, check if the maintenance copies give a full set
-    Map<Integer, Integer> healthy = new HashMap<>(healthyIndexes);
+    // Check if the maintenance copies give a full set and also that we do not
+    // have too many in maintenance
+    Map<Integer, Integer> healthy = new HashMap<>(onlineIndexes);
     maintenanceIndexes.forEach((k, v) -> healthy.merge(k, v, Integer::sum));
-    return hasFullSetOfIndexes(healthy);
+    return hasFullSetOfIndexes(healthy) && onlineIndexes.size()
+        >= repConfig.getData() + remainingMaintenanceRedundancy;
   }
 
   /**

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

@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership.  The ASF
+ * licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package org.apache.hadoop.hdds.scm.container.replication;
+
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.scm.container.ContainerInfo;
+import org.apache.hadoop.hdds.scm.container.ContainerReplica;
+
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Interface used by ReplicationManager to check if containers are healthy or
+ * not.
+ */
+public interface ContainerHealthCheck {
+
+  ContainerHealthResult checkHealth(
+      ContainerInfo container, Set<ContainerReplica> replicas,
+      List<Pair<Integer, DatanodeDetails>> indexesPendingAdd,
+      List<Pair<Integer, DatanodeDetails>> indexesPendingDelete,
+      int remainingRedundancyForMaintenance);
+}