HDDS-6280. Support Container Balancer HA

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

@@ -141,6 +141,7 @@ public ContainerBalancer(StorageContainerManager scm) {
 
     this.lock = new ReentrantLock();
     findSourceStrategy = new FindSourceGreedy(nodeManager);
+    scm.getSCMServiceManager().register(this);
   }
 
   /**
@@ -196,12 +197,8 @@ private void balance() {
           return;
         }
         // otherwise, try to stop balancer
-        try {
-          stopBalancer();
-        } catch (IOException | IllegalContainerBalancerStateException e) {
-          LOG.warn("Tried and failed to stop Container Balancer when it " +
-              "could not initialize an iteration", e);
-        }
+        tryStopBalancer("Could not initialize ContainerBalancer's " +
+            "iteration number " + i);
         return;
       }
 
@@ -211,30 +208,18 @@ private void balance() {
 
       // persist next iteration index
       if (iR == IterationResult.ITERATION_COMPLETED) {
-        lock.lock();
         try {
-          saveConfiguration(
-              config.toProtobufBuilder()
-                  .setShouldRun(true)
-                  .setNextIterationIndex(i + 1)
-                  .build());
+          saveConfiguration(config, true, i + 1);
         } catch (IOException e) {
           LOG.warn("Could not persist next iteration index value for " +
               "ContainerBalancer after completing an iteration", e);
-        } finally {
-          lock.unlock();
         }
       }
 
       // if no new move option is generated, it means the cluster cannot be
       // balanced anymore; so just stop balancer
       if (iR == IterationResult.CAN_NOT_BALANCE_ANY_MORE) {
-        try {
-          stopBalancer();
-        } catch (IOException | IllegalContainerBalancerStateException e) {
-          LOG.warn("Tried and failed to stop Container Balancer when result " +
-              "of the latest iteration was " + iR, e);
-        }
+        tryStopBalancer(iR.toString());
         return;
       }
 
@@ -260,13 +245,8 @@ private void balance() {
     }
 
     // finally, stop balancer if it hasn't been stopped already
-    try {
-      if (isBalancerRunning()) {
-        stopBalancer();
-      }
-    } catch (IOException | IllegalContainerBalancerStateException e) {
-      LOG.warn("Failed to stop Container Balancer after it completed all " +
-          "iterations", e);
+    if (isBalancerRunning()) {
+      tryStopBalancer("Completed all iterations.");
     }
   }
 
@@ -279,7 +259,7 @@ private void balance() {
    */
   private boolean initializeIteration() {
     if (scmContext.isInSafeMode()) {
-      LOG.warn("Container Balancer cannot operate while SCM is in Safe Mode.");
+      LOG.error("Container Balancer cannot operate while SCM is in Safe Mode.");
       return false;
     }
     if (!scmContext.isLeader()) {
@@ -861,11 +841,13 @@ public void notifyStatusChanged() {
     try {
       if (!scmContext.isLeader() || scmContext.isInSafeMode()) {
         if (isBalancerRunning()) {
+          LOG.info("Stopping ContainerBalancer in this scm on status change");
           stop();
         }
       } else {
         if (shouldRun()) {
           try {
+            LOG.info("Starting ContainerBalancer in this scm on status change");
             start();
           } catch (IllegalContainerBalancerStateException |
               InvalidContainerBalancerConfigurationException e) {
@@ -1071,16 +1053,33 @@ public void stopBalancer()
     try {
       // should be leader, out of safe mode, and currently running
       validateState(true);
-      saveConfiguration(config.toProtobufBuilder()
-          .setShouldRun(false)
-          .setNextIterationIndex(0)
-          .build());
+      saveConfiguration(config, false, 0);
       stop();
     } finally {
       lock.unlock();
     }
   }
 
+  /**
+   * Tries to stop ContainerBalancer. Logs the reason for stopping. Calls
+   * {@link ContainerBalancer#stopBalancer()}.
+   * @param stopReason a string specifying the reason for stopping
+   *                   ContainerBalancer.
+   */
+  private void tryStopBalancer(String stopReason) {
+    lock.lock();
+    try {
+      LOG.info("Stopping ContainerBalancer. Reason for stopping: {}",
+          stopReason);
+      stopBalancer();
+    } catch (IllegalContainerBalancerStateException | IOException e) {
+      LOG.warn("Tried to stop ContainerBalancer but failed. Reason for " +
+          "stopping: {}", stopReason, e);
+    } finally {
+      lock.unlock();
+    }
+  }
+
   private void stopBalancingThread() {
     Thread balancingThread;
     lock.lock();
@@ -1103,6 +1102,20 @@ private void stopBalancingThread() {
     LOG.info("Container Balancer stopped successfully.");
   }
 
+  private void saveConfiguration(ContainerBalancerConfiguration configuration,
+                                 boolean shouldRun, int index)
+      throws IOException {
+    lock.lock();
+    try {
+      saveConfiguration(configuration.toProtobufBuilder()
+          .setShouldRun(shouldRun)
+          .setNextIterationIndex(index)
+          .build());
+    } finally {
+      lock.unlock();
+    }
+  }
+
   private void validateConfiguration(ContainerBalancerConfiguration conf)
       throws InvalidContainerBalancerConfigurationException {
     // maxSizeEnteringTarget and maxSizeLeavingSource should by default be
@@ -1158,27 +1171,24 @@ public void setOzoneConfiguration(
   }
 
   /**
-   * Persists the configuration that ContainerBalancer will use after validating
-   * state and the specified configuration.
-   *  @param configuration ContainerBalancerConfiguration to persist
-   *
+   * Persists the configuration that ContainerBalancer will use after
+   * validating state and the specified configuration.
+   * @param configuration ContainerBalancerConfiguration to persist
+   * @throws InvalidContainerBalancerConfigurationException on failure to
+   * validate the specified configuration
+   * @throws IllegalContainerBalancerStateException if this SCM is not leader
+   * or not out of safe mode or if ContainerBalancer is currently running in
+   * this SCM
+   * @throws IOException on failure to persist configuration
    */
   private void setBalancerConfigOnStartBalancer(
       ContainerBalancerConfiguration configuration)
       throws InvalidContainerBalancerConfigurationException,
       IllegalContainerBalancerStateException, IOException {
     validateState(false);
     validateConfiguration(configuration);
-    lock.lock();
-    try {
-      saveConfiguration(configuration.toProtobufBuilder()
-          .setShouldRun(true)
-          .setNextIterationIndex(0)
-          .build());
-      this.config = configuration;
-    } finally {
-      lock.unlock();
-    }
+    saveConfiguration(configuration, true, 0);
+    this.config = configuration;
   }
 
   /**

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

@@ -134,8 +134,6 @@ public final class ContainerBalancerConfiguration {
           "data node is very high")
   private boolean triggerDuEnable = false;
 
-  private int nextIterationIndex = 0;
-
   /**
    * Gets the threshold value for Container Balancer.
    *

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/StatefulServiceStateManagerImpl.java

@@ -23,6 +23,8 @@
 import org.apache.hadoop.hdds.protocol.proto.SCMRatisProtocol.RequestType;
 import org.apache.hadoop.hdds.scm.metadata.DBTransactionBuffer;
 import org.apache.hadoop.hdds.utils.db.Table;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.lang.reflect.Proxy;
@@ -34,6 +36,9 @@
 public final class StatefulServiceStateManagerImpl
     implements StatefulServiceStateManager {
 
+  public static final Logger LOG =
+      LoggerFactory.getLogger(StatefulServiceStateManagerImpl.class);
+
   // this table maps the service name to the configuration (ByteString)
   private Table<String, ByteString> statefulServiceConfig;
   private final DBTransactionBuffer transactionBuffer;
@@ -52,10 +57,13 @@ private StatefulServiceStateManagerImpl(
   public void saveConfiguration(String serviceName, ByteString bytes)
       throws IOException {
     transactionBuffer.addToBuffer(statefulServiceConfig, serviceName, bytes);
+    LOG.debug("Added specified bytes to the transaction buffer for key {} to " +
+        "table {}", serviceName, statefulServiceConfig.getName());
     if (transactionBuffer instanceof SCMHADBTransactionBuffer) {
       SCMHADBTransactionBuffer buffer =
               (SCMHADBTransactionBuffer) transactionBuffer;
       buffer.flush();
+      LOG.debug("Transaction buffer flushed");
     }
   }
 

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/io/ByteStringCodec.java

@@ -0,0 +1,39 @@
+/*
+ * 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.ha.io;
+
+import com.google.protobuf.ByteString;
+import com.google.protobuf.InvalidProtocolBufferException;
+
+/**
+ * A dummy codec that serializes a ByteString object to ByteString.
+ */
+public class ByteStringCodec implements Codec {
+
+  @Override
+  public ByteString serialize(Object object)
+      throws InvalidProtocolBufferException {
+    return (ByteString) object;
+  }
+
+  @Override
+  public Object deserialize(Class<?> type, ByteString value)
+      throws InvalidProtocolBufferException {
+    return value;
+  }
+}

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/io/CodecFactory.java

@@ -17,6 +17,7 @@
 
 package org.apache.hadoop.hdds.scm.ha.io;
 
+import com.google.protobuf.ByteString;
 import com.google.protobuf.GeneratedMessage;
 import com.google.protobuf.InvalidProtocolBufferException;
 import com.google.protobuf.ProtocolMessageEnum;
@@ -45,6 +46,7 @@ public final class CodecFactory {
     codecs.put(Boolean.class, new BooleanCodec());
     codecs.put(BigInteger.class, new BigIntegerCodec());
     codecs.put(X509Certificate.class, new X509CertificateCodec());
+    codecs.put(ByteString.class, new ByteStringCodec());
   }
 
   private CodecFactory() { }

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

@@ -1471,7 +1471,8 @@ public void stop() {
     try {
       if (containerBalancer.isBalancerRunning()) {
         LOG.info("Stopping Container Balancer service.");
-        containerBalancer.stopBalancer();
+        // stop ContainerBalancer thread in this scm
+        containerBalancer.stop();
       } else {
         LOG.info("Container Balancer is not running.");
       }

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/balancer/TestContainerBalancer.java

@@ -39,6 +39,8 @@
 import org.apache.hadoop.hdds.scm.container.placement.algorithms.SCMContainerPlacementMetrics;
 import org.apache.hadoop.hdds.scm.container.placement.metrics.SCMNodeStat;
 import org.apache.hadoop.hdds.scm.ha.SCMContext;
+import org.apache.hadoop.hdds.scm.ha.SCMService;
+import org.apache.hadoop.hdds.scm.ha.SCMServiceManager;
 import org.apache.hadoop.hdds.scm.ha.StatefulServiceStateManager;
 import org.apache.hadoop.hdds.scm.ha.StatefulServiceStateManagerImpl;
 import org.apache.hadoop.hdds.scm.node.DatanodeUsageInfo;
@@ -111,6 +113,7 @@ public void setup() throws IOException, NodeNotFoundException {
     containerManager = Mockito.mock(ContainerManager.class);
     replicationManager = Mockito.mock(ReplicationManager.class);
     serviceStateManager = Mockito.mock(StatefulServiceStateManagerImpl.class);
+    SCMServiceManager scmServiceManager = Mockito.mock(SCMServiceManager.class);
 
     // these configs will usually be specified in each test
     balancerConfiguration =
@@ -164,6 +167,7 @@ public void setup() throws IOException, NodeNotFoundException {
     when(scm.getEventQueue()).thenReturn(mock(EventPublisher.class));
     when(scm.getConfiguration()).thenReturn(conf);
     when(scm.getStatefulServiceStateManager()).thenReturn(serviceStateManager);
+    when(scm.getSCMServiceManager()).thenReturn(scmServiceManager);
 
     /*
     When StatefulServiceStateManager#saveConfiguration is called, save to
@@ -184,6 +188,8 @@ public void setup() throws IOException, NodeNotFoundException {
     when(serviceStateManager.readConfiguration(Mockito.anyString())).thenAnswer(
         i -> serviceToConfigMap.get(i.getArgument(0, String.class)));
 
+    Mockito.doNothing().when(scmServiceManager)
+        .register(Mockito.any(SCMService.class));
     containerBalancer = new ContainerBalancer(scm);
   }