HDDS-10372. SCM and Datanode communication for reconciliation

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/client/ScmClient.java

@@ -462,4 +462,12 @@ DecommissionScmResponseProto decommissionScm(
       String scmId) throws IOException;
 
   String getMetrics(String query) throws IOException;
+
+  /**
+   * Trigger a reconcile command to datanodes for a container ID.
+   *
+   * @param containerID The ID of the container to reconcile.
+   * @throws IOException On error
+   */
+  void reconcileContainer(long containerID) throws IOException;
 }

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerReplicaInfo.java

@@ -17,9 +17,14 @@
  */
 package org.apache.hadoop.hdds.scm.container;
 
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 
+import java.io.IOException;
 import java.util.UUID;
 
 /**
@@ -35,6 +40,8 @@ public final class ContainerReplicaInfo {
   private long keyCount;
   private long bytesUsed;
   private int replicaIndex = -1;
+  @JsonSerialize(using = LongToHexJsonSerializer.class)
+  private long dataChecksum;
 
   public static ContainerReplicaInfo fromProto(
       HddsProtos.SCMContainerReplicaProto proto) {
@@ -48,7 +55,8 @@ public static ContainerReplicaInfo fromProto(
         .setKeyCount(proto.getKeyCount())
         .setBytesUsed(proto.getBytesUsed())
         .setReplicaIndex(
-            proto.hasReplicaIndex() ? (int)proto.getReplicaIndex() : -1);
+            proto.hasReplicaIndex() ? (int)proto.getReplicaIndex() : -1)
+        .setDataChecksum(proto.getDataChecksum());
     return builder.build();
   }
 
@@ -87,6 +95,17 @@ public int getReplicaIndex() {
     return replicaIndex;
   }
 
+  public long getDataChecksum() {
+    return dataChecksum;
+  }
+
+  private static class LongToHexJsonSerializer extends JsonSerializer<Long> {
+    @Override
+    public void serialize(Long value, JsonGenerator gen, SerializerProvider provider) throws IOException {
+      gen.writeString(Long.toHexString(value));
+    }
+  }
+
   /**
    * Builder for ContainerReplicaInfo class.
    */
@@ -134,6 +153,11 @@ public Builder setReplicaIndex(int replicaIndex) {
       return this;
     }
 
+    public Builder setDataChecksum(long dataChecksum) {
+      subject.dataChecksum = dataChecksum;
+      return this;
+    }
+
     public ContainerReplicaInfo build() {
       return subject;
     }

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/exceptions/SCMException.java

@@ -139,6 +139,7 @@ public enum ResultCodes {
     CA_ROTATION_IN_PROGRESS,
     CA_ROTATION_IN_POST_PROGRESS,
     CONTAINER_ALREADY_CLOSED,
-    CONTAINER_ALREADY_CLOSING
+    CONTAINER_ALREADY_CLOSING,
+    UNSUPPORTED_OPERATION
   }
 }

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/protocol/StorageContainerLocationProtocol.java

@@ -483,4 +483,12 @@ DecommissionScmResponseProto decommissionScm(
       String scmId) throws IOException;
 
   String getMetrics(String query) throws IOException;
+
+  /**
+   * Trigger a reconcile command to datanodes for the current container ID.
+   *
+   * @param containerID The ID of the container to reconcile.
+   * @throws IOException On error
+   */
+  void reconcileContainer(long containerID) throws IOException;
 }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ContainerUtils.java

@@ -194,21 +194,21 @@ public static synchronized DatanodeDetails readDatanodeDetailsFrom(File path)
    * Verify that the checksum stored in containerData is equal to the
    * computed checksum.
    */
-  public static void verifyChecksum(ContainerData containerData,
+  public static void verifyContainerFileChecksum(ContainerData containerData,
       ConfigurationSource conf) throws IOException {
     boolean enabled = conf.getBoolean(
             HddsConfigKeys.HDDS_CONTAINER_CHECKSUM_VERIFICATION_ENABLED,
             HddsConfigKeys.
                     HDDS_CONTAINER_CHECKSUM_VERIFICATION_ENABLED_DEFAULT);
     if (enabled) {
-      String storedChecksum = containerData.getChecksum();
+      String storedChecksum = containerData.getContainerFileChecksum();
 
       Yaml yaml = ContainerDataYaml.getYamlForContainerType(
           containerData.getContainerType(),
           containerData instanceof KeyValueContainerData &&
               ((KeyValueContainerData)containerData).getReplicaIndex() > 0);
-      containerData.computeAndSetChecksum(yaml);
-      String computedChecksum = containerData.getChecksum();
+      containerData.computeAndSetContainerFileChecksum(yaml);
+      String computedChecksum = containerData.getContainerFileChecksum();
 
       if (storedChecksum == null || !storedChecksum.equals(computedChecksum)) {
         throw new StorageContainerException("Container checksum error for " +
@@ -225,7 +225,7 @@ public static void verifyChecksum(ContainerData containerData,
    * @param containerDataYamlStr ContainerData as a Yaml String
    * @return Checksum of the container data
    */
-  public static String getChecksum(String containerDataYamlStr)
+  public static String getContainerFileChecksum(String containerDataYamlStr)
       throws StorageContainerException {
     MessageDigest sha;
     try {

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

@@ -99,8 +99,12 @@ public abstract class ContainerData {
 
   private HddsVolume volume;
 
+  // Checksum of just the container file.
   private String checksum;
 
+  // Checksum of the data within the container.
+  private long dataChecksum;
+
   private boolean isEmpty;
 
   private int replicaIndex;
@@ -112,7 +116,7 @@ public abstract class ContainerData {
   private transient Optional<Instant> lastDataScanTime = Optional.empty();
 
   public static final Charset CHARSET_ENCODING = StandardCharsets.UTF_8;
-  private static final String DUMMY_CHECKSUM = new String(new byte[64],
+  private static final String ZERO_CHECKSUM = new String(new byte[64],
       CHARSET_ENCODING);
 
   // Common Fields need to be stored in .container file.
@@ -159,7 +163,8 @@ protected ContainerData(ContainerType type, long containerId,
     this.originPipelineId = originPipelineId;
     this.originNodeId = originNodeId;
     this.isEmpty = false;
-    setChecksumTo0ByteArray();
+    this.checksum = ZERO_CHECKSUM;
+    this.dataChecksum = 0;
   }
 
   protected ContainerData(ContainerData source) {
@@ -571,15 +576,11 @@ public void setBlockCount(long count) {
     this.blockCount.set(count);
   }
 
-  public void setChecksumTo0ByteArray() {
-    this.checksum = DUMMY_CHECKSUM;
-  }
-
-  public void setChecksum(String checkSum) {
+  public void setContainerFileChecksum(String checkSum) {
     this.checksum = checkSum;
   }
 
-  public String getChecksum() {
+  public String getContainerFileChecksum() {
     return this.checksum;
   }
 
@@ -630,21 +631,29 @@ public String getOriginNodeId() {
    *
    * Checksum of ContainerData is calculated by setting the
    * {@link ContainerData#checksum} field to a 64-byte array with all 0's -
-   * {@link ContainerData#DUMMY_CHECKSUM}. After the checksum is calculated,
+   * {@link ContainerData#ZERO_CHECKSUM}. After the checksum is calculated,
    * the checksum field is updated with this value.
    *
    * @param yaml Yaml for ContainerType to get the ContainerData as Yaml String
    * @throws IOException
    */
-  public void computeAndSetChecksum(Yaml yaml) throws IOException {
+  public void computeAndSetContainerFileChecksum(Yaml yaml) throws IOException {
     // Set checksum to dummy value - 0 byte array, to calculate the checksum
     // of rest of the data.
-    setChecksumTo0ByteArray();
+    this.checksum = ZERO_CHECKSUM;
 
     // Dump yaml data into a string to compute its checksum
     String containerDataYamlStr = yaml.dump(this);
 
-    this.checksum = ContainerUtils.getChecksum(containerDataYamlStr);
+    this.checksum = ContainerUtils.getContainerFileChecksum(containerDataYamlStr);
+  }
+
+  public void setDataChecksum(long dataChecksum) {
+    this.dataChecksum = dataChecksum;
+  }
+
+  public long getDataChecksum() {
+    return dataChecksum;
   }
 
   /**

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

@@ -98,7 +98,7 @@ public static void createContainerFile(ContainerType containerType,
       // Create Yaml for given container type
       Yaml yaml = getYamlForContainerType(containerType, withReplicaIndex);
       // Compute Checksum and update ContainerData
-      containerData.computeAndSetChecksum(yaml);
+      containerData.computeAndSetContainerFileChecksum(yaml);
 
       // Write the ContainerData with checksum to Yaml file.
       out = new FileOutputStream(
@@ -312,7 +312,7 @@ public Object construct(Node node) {
         kvData.setChunksPath((String) nodes.get(OzoneConsts.CHUNKS_PATH));
         Map<String, String> meta = (Map) nodes.get(OzoneConsts.METADATA);
         kvData.setMetadata(meta);
-        kvData.setChecksum((String) nodes.get(OzoneConsts.CHECKSUM));
+        kvData.setContainerFileChecksum((String) nodes.get(OzoneConsts.CHECKSUM));
         Long timestamp = (Long) nodes.get(OzoneConsts.DATA_SCAN_TIMESTAMP);
         kvData.setDataScanTimestamp(timestamp);
         String state = (String) nodes.get(OzoneConsts.STATE);

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

@@ -21,8 +21,10 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.util.List;
 
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 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;
@@ -192,6 +194,13 @@ public abstract void closeContainer(Container container)
   public abstract void deleteContainer(Container container, boolean force)
       throws IOException;
 
+  /**
+   * Triggers reconciliation of this container replica's data with its peers.
+   * @param container container to be reconciled.
+   * @param peers The other datanodes with a copy of this container whose data should be checked.
+   */
+  public abstract void reconcileContainer(Container<?> container, List<DatanodeDetails> peers) throws IOException;
+
   /**
    * Deletes the given files associated with a block of the container.
    *

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java

@@ -56,6 +56,7 @@
 import org.apache.hadoop.ozone.container.common.statemachine.commandhandler.DeleteBlocksCommandHandler;
 import org.apache.hadoop.ozone.container.common.statemachine.commandhandler.DeleteContainerCommandHandler;
 import org.apache.hadoop.ozone.container.common.statemachine.commandhandler.FinalizeNewLayoutVersionCommandHandler;
+import org.apache.hadoop.ozone.container.common.statemachine.commandhandler.ReconcileContainerCommandHandler;
 import org.apache.hadoop.ozone.container.common.statemachine.commandhandler.ReconstructECContainersCommandHandler;
 import org.apache.hadoop.ozone.container.common.statemachine.commandhandler.RefreshVolumeUsageCommandHandler;
 import org.apache.hadoop.ozone.container.common.statemachine.commandhandler.ReplicateContainerCommandHandler;
@@ -258,6 +259,7 @@ public DatanodeStateMachine(DatanodeDetails datanodeDetails,
             supervisor::nodeStateUpdated))
         .addHandler(new FinalizeNewLayoutVersionCommandHandler())
         .addHandler(new RefreshVolumeUsageCommandHandler())
+        .addHandler(new ReconcileContainerCommandHandler(threadNamePrefix))
         .setConnectionManager(connectionManager)
         .setContainer(container)
         .setContext(context)

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/ReconcileContainerCommandHandler.java

@@ -0,0 +1,110 @@
+/*
+ * 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.ozone.container.common.statemachine.commandhandler;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto;
+import org.apache.hadoop.ozone.container.common.statemachine.SCMConnectionManager;
+import org.apache.hadoop.ozone.container.common.statemachine.StateContext;
+import org.apache.hadoop.ozone.container.ozoneimpl.OzoneContainer;
+import org.apache.hadoop.ozone.protocol.commands.ReconcileContainerCommand;
+import org.apache.hadoop.ozone.protocol.commands.SCMCommand;
+import org.apache.hadoop.util.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Handles commands from SCM to reconcile a container replica on this datanode with the replicas on its peers.
+ */
+public class ReconcileContainerCommandHandler implements CommandHandler {
+  private static final Logger LOG =
+      LoggerFactory.getLogger(ReconcileContainerCommandHandler.class);
+
+  private final AtomicLong invocationCount;
+  private final AtomicInteger queuedCount;
+  private final ExecutorService executor;
+  private long totalTime;
+
+  public ReconcileContainerCommandHandler(String threadNamePrefix) {
+    invocationCount = new AtomicLong(0);
+    queuedCount = new AtomicInteger(0);
+    // TODO Allow configurable thread pool size with a default value when the implementation is ready.
+    executor = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder()
+        .setNameFormat(threadNamePrefix + "ReconcileContainerThread-%d")
+        .build());
+    totalTime = 0;
+  }
+
+  @Override
+  public void handle(SCMCommand command, OzoneContainer container, StateContext context,
+      SCMConnectionManager connectionManager) {
+    queuedCount.incrementAndGet();
+    CompletableFuture.runAsync(() -> {
+      invocationCount.incrementAndGet();
+      long startTime = Time.monotonicNow();
+      ReconcileContainerCommand reconcileCommand = (ReconcileContainerCommand) command;
+      LOG.info("Processing reconcile container command for container {} with peers {}",
+          reconcileCommand.getContainerID(), reconcileCommand.getPeerDatanodes());
+      try {
+        container.getController().reconcileContainer(reconcileCommand.getContainerID(),
+            reconcileCommand.getPeerDatanodes());
+      } catch (IOException ex) {
+        LOG.error("Failed to reconcile container {}.", reconcileCommand.getContainerID(), ex);
+      } finally {
+        long endTime = Time.monotonicNow();
+        totalTime += endTime - startTime;
+      }
+    }, executor).whenComplete((v, e) -> queuedCount.decrementAndGet());
+  }
+
+  @Override
+  public SCMCommandProto.Type getCommandType() {
+    return SCMCommandProto.Type.reconcileContainerCommand;
+  }
+
+  @Override
+  public int getInvocationCount() {
+    return (int)invocationCount.get();
+  }
+
+  @Override
+  public long getAverageRunTime() {
+    if (invocationCount.get() > 0) {
+      return totalTime / invocationCount.get();
+    }
+    return 0;
+  }
+
+  @Override
+  public long getTotalRunTime() {
+    return totalTime;
+  }
+
+  @Override
+  public int getQueuedCount() {
+    return queuedCount.get();
+  }
+}