HDDS-10928. Implement container comparison logic within datanodes.

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/checksum/ContainerChecksumTreeManager.java

@@ -16,6 +16,7 @@
  */
 package org.apache.hadoop.ozone.container.checksum;
 
+import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
 import com.google.common.annotations.VisibleForTesting;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
@@ -30,12 +31,16 @@
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.Collection;
 import java.util.SortedSet;
 import java.util.TreeSet;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
 
 import com.google.common.util.concurrent.Striped;
 import org.apache.hadoop.hdds.utils.SimpleStriped;
@@ -143,12 +148,81 @@ public void markBlocksAsDeleted(KeyValueContainerData data, Collection<Long> del
     }
   }
 
-  public ContainerDiff diff(KeyValueContainerData thisContainer, ContainerProtos.ContainerChecksumInfo otherInfo)
+  public ContainerDiff diff(KeyValueContainerData thisContainer, ContainerProtos.ContainerChecksumInfo peerChecksumInfo)
       throws IOException {
-    // TODO HDDS-10928 compare the checksum info of the two containers and return a summary.
-    //  Callers can act on this summary to repair their container replica using the peer's replica.
-    //  This method will use the read lock, which is unused in the current implementation.
-    return new ContainerDiff();
+    Preconditions.assertNotNull(thisContainer, "Container data is null");
+    Optional<ContainerProtos.ContainerChecksumInfo.Builder> thisContainerChecksumInfoBuilder =
+        read(thisContainer);
+    if (!thisContainerChecksumInfoBuilder.isPresent()) {
+      // TODO: To create containerMerkleTree or fail the request.
+      return null;
+    }
+
+    if (thisContainer.getContainerID() != peerChecksumInfo.getContainerID()) {
+      throw new IOException("Container Id does not match");
+    }
+
+    ContainerProtos.ContainerChecksumInfo thisChecksumInfo = thisContainerChecksumInfoBuilder.get().build();
+
+    ContainerProtos.ContainerMerkleTree thisMerkleTree = thisChecksumInfo.getContainerMerkleTree();
+    ContainerProtos.ContainerMerkleTree peerMerkleTree = peerChecksumInfo.getContainerMerkleTree();
+
+    return compareContainerMerkleTree(thisMerkleTree, peerMerkleTree);
+  }
+
+  private ContainerDiff compareContainerMerkleTree(ContainerProtos.ContainerMerkleTree thisMerkleTree,
+                                                   ContainerProtos.ContainerMerkleTree peerMerkleTree) {
+
+    ContainerDiff report = new ContainerDiff();
+    if (thisMerkleTree.getDataChecksum() == peerMerkleTree.getDataChecksum()) {
+      return new ContainerDiff();
+    }
+
+    Map<Long, ContainerProtos.BlockMerkleTree> thisBlockMerkleTreeMap = thisMerkleTree.getBlockMerkleTreeList()
+        .stream().collect(Collectors.toMap(ContainerProtos.BlockMerkleTree::getBlockID,
+            blockMerkleTree -> blockMerkleTree));
+
+    // Since we are reconciling our container with a peer, We only need to go through the peer's block list
+    for (ContainerProtos.BlockMerkleTree peerBlockMerkleTree: peerMerkleTree.getBlockMerkleTreeList()) {
+      // Check if our container has the peer block.
+      ContainerProtos.BlockMerkleTree thisBlockMerkleTree = thisBlockMerkleTreeMap.get(
+          peerBlockMerkleTree.getBlockID());
+      if (thisBlockMerkleTree == null) {
+        report.addMissingBlock(peerBlockMerkleTree);
+        continue;
+      }
+
+      if (thisBlockMerkleTree.getBlockChecksum() != peerBlockMerkleTree.getBlockChecksum()) {
+        compareBlockMerkleTree(report, thisBlockMerkleTree, peerBlockMerkleTree);
+      }
+    }
+    return report;
+  }
+
+  private void compareBlockMerkleTree(ContainerDiff report, ContainerProtos.BlockMerkleTree thisBlockMerkleTree,
+                                      ContainerProtos.BlockMerkleTree peerBlockMerkleTree) {
+    Map<Long, ContainerProtos.ChunkMerkleTree> thisChunkMerkleTreeMap = thisBlockMerkleTree.getChunkMerkleTreeList()
+        .stream().collect(Collectors.toMap(ContainerProtos.ChunkMerkleTree::getOffset,
+            chunkMerkleTree -> chunkMerkleTree));
+    List<ContainerProtos.ChunkMerkleTree> peerChunkMerkleTreeList = peerBlockMerkleTree.getChunkMerkleTreeList();
+
+    // Since we are reconciling our container with a peer, We only need to go through the peer's chunk list
+    for (ContainerProtos.ChunkMerkleTree peerChunkMerkleTree: peerChunkMerkleTreeList) {
+      ContainerProtos.ChunkMerkleTree thisChunkMerkleTree = thisChunkMerkleTreeMap.get(
+          peerChunkMerkleTree.getOffset());
+      if (thisChunkMerkleTree == null) {
+        report.addMissingChunk(peerChunkMerkleTree);
+        continue;
+      }
+
+      if (thisChunkMerkleTree.getChunkChecksum() != peerChunkMerkleTree.getChunkChecksum()) {
+        ChunkArgs thisChunkArgs = new ChunkArgs(thisBlockMerkleTree.getBlockID(), thisChunkMerkleTree.getOffset(),
+            thisChunkMerkleTree.getLength(), thisChunkMerkleTree.getChunkChecksum());
+        ChunkArgs peerChunkArgs = new ChunkArgs(peerBlockMerkleTree.getBlockID(), peerChunkMerkleTree.getOffset(),
+            peerChunkMerkleTree.getLength(), peerChunkMerkleTree.getChunkChecksum());
+        report.addCorruptChunks(Pair.of(thisChunkArgs, peerChunkArgs));
+      }
+    }
   }
 
   /**
@@ -245,11 +319,73 @@ public static boolean checksumFileExist(Container container) {
    * This class represents the difference between our replica of a container and a peer's replica of a container.
    * It summarizes the operations we need to do to reconcile our replica with the peer replica it was compared to.
    *
-   * TODO HDDS-10928
    */
   public static class ContainerDiff {
+    private final List<ContainerProtos.BlockMerkleTree> missingBlocks;
+    private final List<ContainerProtos.ChunkMerkleTree> missingChunks;
+    private final List<Pair<ChunkArgs, ChunkArgs>> corruptChunks;
+
     public ContainerDiff() {
+      this.missingBlocks = new ArrayList<>();
+      this.missingChunks = new ArrayList<>();
+      this.corruptChunks = new ArrayList<>();
+    }
+
+    public void addMissingBlock(ContainerProtos.BlockMerkleTree otherBlockMerkleTree) {
+      this.missingBlocks.add(otherBlockMerkleTree);
+    }
+
+    public void addMissingChunk(ContainerProtos.ChunkMerkleTree otherChunkMerkleTree) {
+      this.missingChunks.add(otherChunkMerkleTree);
+    }
+
+    public void addCorruptChunks(Pair<ChunkArgs, ChunkArgs> mismatchedChunk) {
+      this.corruptChunks.add(mismatchedChunk);
+    }
+
+    public List<ContainerProtos.BlockMerkleTree> getMissingBlocks() {
+      return missingBlocks;
+    }
+
+    public List<ContainerProtos.ChunkMerkleTree> getMissingChunks() {
+      return missingChunks;
+    }
+
+    public List<Pair<ChunkArgs, ChunkArgs>> getCorruptChunks() {
+      return corruptChunks;
+    }
+  }
+
+  /**
+   * This class encapsulates chunk merkle tree with block ID which is required for reconciliation.
+   */
+  public static class ChunkArgs {
+    private final long blockId;
+    private final long chunkOffset;
+    private final long chunkLength;
+    private final long checksum;
+
+    public ChunkArgs(long blockId, long chunkOffset, long chunkLength, long checksum) {
+      this.blockId = blockId;
+      this.chunkOffset = chunkOffset;
+      this.chunkLength = chunkLength;
+      this.checksum = checksum;
+    }
+
+    public long getBlockId() {
+      return blockId;
+    }
+
+    public long getChunkOffset() {
+      return chunkOffset;
+    }
+
+    public long getChunkLength() {
+      return chunkLength;
+    }
 
+    public long getChecksum() {
+      return checksum;
     }
   }
 }

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/checksum/ContainerMerkleTreeTestUtils.java

@@ -146,6 +146,40 @@ public static ContainerMerkleTree buildTestTree(ConfigurationSource conf) {
     return tree;
   }
 
+  public static ContainerMerkleTree buildTestTreeWithMismatches(ConfigurationSource conf, boolean missingBlocks,
+                                                                boolean missingChunks, boolean corruptChunks) {
+    final long blockID1 = 1;
+    final long blockID2 = 2;
+    final long blockID3 = 3;
+    ContainerProtos.ChunkInfo b1c1 = buildChunk(conf, 0, ByteBuffer.wrap(new byte[]{1, 2, 3}));
+    ContainerProtos.ChunkInfo b1c2 = buildChunk(conf, 1, ByteBuffer.wrap(new byte[]{4, 5, 6}));
+    ContainerProtos.ChunkInfo b2c1 = buildChunk(conf, 0, ByteBuffer.wrap(new byte[]{7, 8, 9}));
+    ContainerProtos.ChunkInfo b2c2 = buildChunk(conf, 1, ByteBuffer.wrap(new byte[]{12, 11, 10}));
+    ContainerProtos.ChunkInfo b3c1 = buildChunk(conf, 0, ByteBuffer.wrap(new byte[]{13, 14, 15}));
+    ContainerProtos.ChunkInfo b3c2 = buildChunk(conf, 1, ByteBuffer.wrap(new byte[]{16, 17, 18}));
+
+    ContainerMerkleTree tree = new ContainerMerkleTree();
+    tree.addChunks(blockID1, Arrays.asList(b1c1, b1c2));
+
+    if (corruptChunks) {
+      ContainerProtos.ChunkInfo corruptB1c2 = buildChunk(conf, 1, ByteBuffer.wrap(new byte[]{2, 4, 6}));
+      tree.addChunks(blockID1, Arrays.asList(b1c1, corruptB1c2));
+    } else {
+      tree.addChunks(blockID1, Arrays.asList(b1c1, b1c2));
+    }
+
+    if (missingChunks) {
+      tree.addChunks(blockID2, Arrays.asList(b2c1));
+    } else {
+      tree.addChunks(blockID2, Arrays.asList(b2c1, b2c2));
+    }
+
+    if (!missingBlocks) {
+      tree.addChunks(blockID3, Arrays.asList(b3c1, b3c2));
+    }
+    return tree;
+  }
+
   /**
    * This function checks whether the container checksum file exists.
    */

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/checksum/TestContainerChecksumTreeManager.java

@@ -20,6 +20,7 @@
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
 import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainerData;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
@@ -37,6 +38,7 @@
 
 import static org.apache.hadoop.ozone.container.checksum.ContainerMerkleTreeTestUtils.assertTreesSortedAndMatch;
 import static org.apache.hadoop.ozone.container.checksum.ContainerMerkleTreeTestUtils.buildTestTree;
+import static org.apache.hadoop.ozone.container.checksum.ContainerMerkleTreeTestUtils.buildTestTreeWithMismatches;
 import static org.apache.hadoop.ozone.container.checksum.ContainerMerkleTreeTestUtils.readChecksumFile;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -299,6 +301,86 @@ public void testEmptyFile() throws Exception {
     assertEquals(CONTAINER_ID, info.getContainerID());
   }
 
+  @Test
+  public void testContainerWithNoDiff() throws IOException {
+    ContainerMerkleTree ourMerkleTree = buildTestTree(config);
+    ContainerMerkleTree peerMerkleTree = buildTestTree(config);
+    checksumManager.writeContainerDataTree(container, ourMerkleTree);
+    ContainerChecksumTreeManager containerChecksumTreeManager = new ContainerChecksumTreeManager(
+        new OzoneConfiguration());
+    ContainerProtos.ContainerChecksumInfo peerChecksumInfo = ContainerProtos.ContainerChecksumInfo.newBuilder()
+            .setContainerID(container.getContainerID())
+            .setContainerMerkleTree(peerMerkleTree.toProto()).build();
+    ContainerChecksumTreeManager.ContainerDiff diff =
+        containerChecksumTreeManager.diff(container, peerChecksumInfo);
+    Assertions.assertTrue(diff.getCorruptChunks().isEmpty());
+    Assertions.assertTrue(diff.getMissingBlocks().isEmpty());
+    Assertions.assertTrue(diff.getMissingChunks().isEmpty());
+  }
+
+  @Test
+  public void testContainerDiffWithMissingBlocksAndChunks() throws IOException {
+    ContainerMerkleTree ourMerkleTree = buildTestTreeWithMismatches(config, true, true, false);
+    ContainerMerkleTree peerMerkleTree = buildTestTree(config);
+    checksumManager.writeContainerDataTree(container, ourMerkleTree);
+    ContainerChecksumTreeManager containerChecksumTreeManager = new ContainerChecksumTreeManager(
+        new OzoneConfiguration());
+    ContainerProtos.ContainerChecksumInfo peerChecksumInfo = ContainerProtos.ContainerChecksumInfo.newBuilder()
+        .setContainerID(container.getContainerID())
+        .setContainerMerkleTree(peerMerkleTree.toProto()).build();
+    ContainerChecksumTreeManager.ContainerDiff diff =
+        containerChecksumTreeManager.diff(container, peerChecksumInfo);
+    Assertions.assertTrue(diff.getCorruptChunks().isEmpty());
+    Assertions.assertFalse(diff.getMissingBlocks().isEmpty());
+    Assertions.assertFalse(diff.getMissingChunks().isEmpty());
+
+    Assertions.assertEquals(diff.getMissingBlocks().size(), 1);
+    Assertions.assertEquals(diff.getMissingChunks().size(), 1);
+  }
+
+  @Test
+  public void testContainerDiffWithMissingBlocksAndMismatchChunks() throws IOException {
+    ContainerMerkleTree ourMerkleTree = buildTestTreeWithMismatches(config, true, false, true);
+    ContainerMerkleTree peerMerkleTree = buildTestTree(config);
+    checksumManager.writeContainerDataTree(container, ourMerkleTree);
+    ContainerChecksumTreeManager containerChecksumTreeManager = new ContainerChecksumTreeManager(
+        new OzoneConfiguration());
+    ContainerProtos.ContainerChecksumInfo peerChecksumInfo = ContainerProtos.ContainerChecksumInfo.newBuilder()
+        .setContainerID(container.getContainerID())
+        .setContainerMerkleTree(peerMerkleTree.toProto()).build();
+    ContainerChecksumTreeManager.ContainerDiff diff =
+        containerChecksumTreeManager.diff(container, peerChecksumInfo);
+    Assertions.assertFalse(diff.getCorruptChunks().isEmpty());
+    Assertions.assertFalse(diff.getMissingBlocks().isEmpty());
+    Assertions.assertTrue(diff.getMissingChunks().isEmpty());
+
+    Assertions.assertEquals(diff.getCorruptChunks().size(), 1);
+    Assertions.assertEquals(diff.getMissingBlocks().size(), 1);
+  }
+
+  /**
+   * Test if a peer which has missing blocks and chunks affects our container diff.
+   * Only if our merkle tree has missing entries from the peer we need to add it the Container Diff.
+   */
+  @Test
+  public void testPeerWithMissingBlockAndMissingChunks() throws IOException {
+    ContainerMerkleTree ourMerkleTree = buildTestTree(config);
+    ContainerMerkleTree peerMerkleTree = buildTestTreeWithMismatches(config, true, true, true);
+    checksumManager.writeContainerDataTree(container, ourMerkleTree);
+    ContainerChecksumTreeManager containerChecksumTreeManager = new ContainerChecksumTreeManager(
+        new OzoneConfiguration());
+    ContainerProtos.ContainerChecksumInfo peerChecksumInfo = ContainerProtos.ContainerChecksumInfo.newBuilder()
+        .setContainerID(container.getContainerID())
+        .setContainerMerkleTree(peerMerkleTree.toProto()).build();
+    ContainerChecksumTreeManager.ContainerDiff diff =
+        containerChecksumTreeManager.diff(container, peerChecksumInfo);
+    Assertions.assertFalse(diff.getCorruptChunks().isEmpty());
+    Assertions.assertTrue(diff.getMissingBlocks().isEmpty());
+    Assertions.assertTrue(diff.getMissingChunks().isEmpty());
+
+    Assertions.assertEquals(diff.getCorruptChunks().size(), 1);
+  }
+
   @Test
   public void testChecksumTreeFilePath() {
     assertEquals(checksumFile.getAbsolutePath(),