Merge branch 'master' into HDDS-13225

Author: YutaLin

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java

@@ -120,6 +120,9 @@ public final class HddsConfigKeys {
   // metadata locations must be configured explicitly.
   public static final String OZONE_METADATA_DIRS = "ozone.metadata.dirs";
 
+  public static final String DATANODE_DB_CONFIG_PATH = "hdds.datanode.db.config.path";
+  public static final String DATANODE_DB_CONFIG_PATH_DEFAULT = "";
+
   public static final String HDDS_PROMETHEUS_ENABLED =
       "hdds.prometheus.endpoint.enabled";
 

hadoop-hdds/common/src/main/resources/ozone-default.xml

@@ -2750,6 +2750,14 @@
       Absolute path to HDDS metadata dir.
     </description>
   </property>
+  <property>
+    <name>hdds.datanode.db.config.path</name>
+    <value/>
+    <tag>OZONE, CONTAINER, STORAGE</tag>
+    <description>
+      Path to an ini configuration file for RocksDB on datanode component.
+    </description>
+  </property>
   <property>
     <name>hdds.priv.key.file.name</name>
     <value>private.pem</value>

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

@@ -31,12 +31,14 @@ public class ContainerDiffReport {
   private final List<ContainerProtos.BlockMerkleTree> missingBlocks;
   private final Map<Long, List<ContainerProtos.ChunkMerkleTree>> missingChunks;
   private final Map<Long, List<ContainerProtos.ChunkMerkleTree>> corruptChunks;
+  private final List<DeletedBlock> divergedDeletedBlocks;
   private final long containerID;
 
   public ContainerDiffReport(long containerID) {
     this.missingBlocks = new ArrayList<>();
     this.missingChunks = new HashMap<>();
     this.corruptChunks = new HashMap<>();
+    this.divergedDeletedBlocks = new ArrayList<>();
     this.containerID = containerID;
   }
 
@@ -67,6 +69,10 @@ public void addCorruptChunk(long blockId, ContainerProtos.ChunkMerkleTree corrup
     this.corruptChunks.computeIfAbsent(blockId, any -> new ArrayList<>()).add(corruptChunk);
   }
 
+  public void addDivergedDeletedBlock(ContainerProtos.BlockMerkleTree blockMerkleTree) {
+    this.divergedDeletedBlocks.add(new DeletedBlock(blockMerkleTree.getBlockID(), blockMerkleTree.getDataChecksum()));
+  }
+
   /**
    * @return A list of BlockMerkleTree objects that were reported as missing.
    */
@@ -88,13 +94,18 @@ public Map<Long, List<ContainerProtos.ChunkMerkleTree>> getCorruptChunks() {
     return corruptChunks;
   }
 
+  public List<DeletedBlock> getDivergedDeletedBlocks() {
+    return divergedDeletedBlocks;
+  }
+
   /**
    * If needRepair is true, It means current replica needs blocks/chunks from the peer to repair
    * its container replica. The peer replica still may have corruption, which it will fix when
    * it reconciles with other peers.
    */
   public boolean needsRepair() {
-    return !missingBlocks.isEmpty() || !missingChunks.isEmpty() || !corruptChunks.isEmpty();
+    return !missingBlocks.isEmpty() || !missingChunks.isEmpty() || !corruptChunks.isEmpty() ||
+        !divergedDeletedBlocks.isEmpty();
   }
 
   public long getNumCorruptChunks() {
@@ -109,11 +120,38 @@ public long getNumMissingBlocks() {
     return missingBlocks.size();
   }
 
+  public long getNumdivergedDeletedBlocks() {
+    return divergedDeletedBlocks.size();
+  }
+
   @Override
   public String toString() {
     return "Diff report for container " + containerID + ":" +
         " Missing Blocks: " + getNumMissingBlocks() +
         " Missing Chunks: " + getNumMissingChunks() + " chunks from " + missingChunks.size() + " blocks" +
-        " Corrupt Chunks: " + getNumCorruptChunks() + " chunks from " + corruptChunks.size() + " blocks";
+        " Corrupt Chunks: " + getNumCorruptChunks() + " chunks from " + corruptChunks.size() + " blocks" +
+        " Diverged Deleted Blocks: " + getNumdivergedDeletedBlocks();
+  }
+
+  /**
+   * Represents a block that has been deleted in a peer whose metadata we need to add to our container replica's
+   * merkle tree.
+   */
+  public static class DeletedBlock {
+    private final long blockID;
+    private final long dataChecksum;
+
+    public DeletedBlock(long blockID, long dataChecksum) {
+      this.blockID = blockID;
+      this.dataChecksum = dataChecksum;
+    }
+
+    public long getBlockID() {
+      return blockID;
+    }
+
+    public long getDataChecksum() {
+      return dataChecksum;
+    }
   }
 }

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

@@ -54,6 +54,9 @@ public class ContainerMerkleTreeMetrics {
   @Metric(about = "Number of corrupt chunks identified during container reconciliation")
   private MutableCounterLong numCorruptChunksIdentified;
 
+  @Metric(about = "Number of diverged block deletes identified during container reconciliation")
+  private MutableCounterLong numDivergedDeletedBlocksIdentified;
+
   @Metric(about = "Merkle tree write latency")
   private MutableRate merkleTreeWriteLatencyNS;
 
@@ -113,6 +116,10 @@ public void incrementCorruptChunksIdentified(long value) {
     this.numCorruptChunksIdentified.incr(value);
   }
 
+  public void incrementDivergedDeletedBlocksIdentified(long value) {
+    this.numDivergedDeletedBlocksIdentified.incr(value);
+  }
+
   public MutableRate getWriteContainerMerkleTreeLatencyNS() {
     return this.merkleTreeWriteLatencyNS;
   }

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

@@ -25,6 +25,7 @@
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
 import org.apache.hadoop.ozone.common.ChecksumByteBuffer;
 import org.apache.hadoop.ozone.common.ChecksumByteBufferFactory;
+import org.apache.hadoop.ozone.container.common.helpers.BlockData;
 import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
 
 /**
@@ -63,9 +64,13 @@ public ContainerMerkleTreeWriter(ContainerProtos.ContainerMerkleTree fromTree) {
     id2Block = new TreeMap<>();
     for (ContainerProtos.BlockMerkleTree blockTree: fromTree.getBlockMerkleTreeList()) {
       long blockID = blockTree.getBlockID();
-      addBlock(blockID);
-      for (ContainerProtos.ChunkMerkleTree chunkTree: blockTree.getChunkMerkleTreeList()) {
-        addChunks(blockID, chunkTree);
+      if (blockTree.getDeleted()) {
+        setDeletedBlock(blockID, blockTree.getDataChecksum());
+      } else {
+        addBlock(blockID);
+        for (ContainerProtos.ChunkMerkleTree chunkTree: blockTree.getChunkMerkleTreeList()) {
+          addChunks(blockID, new ChunkMerkleTreeWriter(chunkTree));
+        }
       }
     }
   }
@@ -81,7 +86,7 @@ public ContainerMerkleTreeWriter(ContainerProtos.ContainerMerkleTree fromTree) {
    */
   public void addChunks(long blockID, boolean checksumMatches, Collection<ContainerProtos.ChunkInfo> chunks) {
     for (ContainerProtos.ChunkInfo chunk: chunks) {
-      addChunks(blockID, checksumMatches, chunk);
+      addChunks(blockID, new ChunkMerkleTreeWriter(chunk, checksumMatches));
     }
   }
 
@@ -91,12 +96,6 @@ public void addChunks(long blockID, boolean checksumMatches, ContainerProtos.Chu
     }
   }
 
-  private void addChunks(long blockID, ContainerProtos.ChunkMerkleTree... chunks) {
-    for (ContainerProtos.ChunkMerkleTree chunkTree: chunks) {
-      addChunks(blockID, new ChunkMerkleTreeWriter(chunkTree));
-    }
-  }
-
   private void addChunks(long blockID, ChunkMerkleTreeWriter chunkWriter) {
     id2Block.computeIfAbsent(blockID, BlockMerkleTreeWriter::new).addChunks(chunkWriter);
   }
@@ -111,13 +110,93 @@ public void addBlock(long blockID) {
     id2Block.computeIfAbsent(blockID, BlockMerkleTreeWriter::new);
   }
 
+  /**
+   * Creates a deleted block entry in the merkle tree and assigns the block this fixed checksum.
+   * If the block already exists with child data it is overwritten.
+   *
+   * This method is used on the reconciliation path to update the data checksum used for a deleted block based on a
+   * peer's value.
+   */
+  public void setDeletedBlock(long blockID, long dataChecksum) {
+    BlockMerkleTreeWriter blockWriter = new BlockMerkleTreeWriter(blockID);
+    blockWriter.markDeleted(dataChecksum);
+    id2Block.put(blockID, blockWriter);
+  }
+
+  /**
+   * Merges the content from the provided tree with this tree writer.
+   * Conflicts where this tree writer and the incoming existingTree parameter have an entry for the same block are
+   * resolved in the following manner:
+   * - A deleted block supersedes a live block
+   *   - Data cannot be un-deleted, so if a delete is ever witnessed, that is the state the block should converge to.
+   * - If both blocks are either deleted or live, the value in this writer supersedes the value in the existingTree
+   * parameter.
+   *   - Our writer has the last witnessed information that is going to be persisted after this merge.
+   *
+   * For example, consider the case where a peer has deleted a block and we have a corrupt copy that has not yet been
+   * deleted. When we reconcile with this peer, we will mark the block as deleted and use the peer's checksum in our
+   * merkle tree to make the trees converge. The "fix" for corrupted data that is supposed to be deleted is to delete
+   * it. After this, if the scanner runs again before the block is deleted, we don't want to update the tree with the
+   * scanner's value because it would again diverge from the peer due to data that is expected to be deleted.
+   * This would cause the checksum to oscillate back and forth until the block is deleted, instead of converging.
+   */
+  public ContainerProtos.ContainerMerkleTree update(ContainerProtos.ContainerMerkleTree existingTree) {
+    for (ContainerProtos.BlockMerkleTree existingBlockTree: existingTree.getBlockMerkleTreeList()) {
+      long blockID = existingBlockTree.getBlockID();
+      BlockMerkleTreeWriter ourBlockTree = id2Block.get(blockID);
+      if (ourBlockTree != null) {
+        // both trees contain the block. We will only consider the incoming/existing value if it does not match our
+        // current state
+        if (!ourBlockTree.isDeleted() && existingBlockTree.getDeleted()) {
+          setDeletedBlock(blockID, existingBlockTree.getDataChecksum());
+        }
+        // In all other cases, keep using our writer's value over the existing one because either:
+        // - The deleted states match between the two blocks OR
+        // - Our block is deleted and the existing one is not, so we have the latest value to use.
+      } else if (existingBlockTree.getDeleted()) {
+        // Our tree does not have this block. Only take the value if it is deleted.
+        // The definitive set of live blocks will come from this tree writer.
+        setDeletedBlock(blockID, existingBlockTree.getDataChecksum());
+      }
+    }
+    return toProtoBuilder().build();
+  }
+
+  /**
+   * Adds deleted blocks to this merkle tree. The blocks' checksums are computed from the checksums in the BlockData.
+   * If a block with the same ID already exists in the tree, it is overwritten as deleted with the checksum computed
+   * from the chunk checksums in the BlockData. If we reconciled with a peer and already marked this block as deleted
+   * during that process, this will overwrite that value. If it changes the block's checksum from what the peer had,
+   * one more round of reconciliation may be required to bring them in sync.
+   *
+   * The top level container data checksum is only computed in the returned tree proto if computeChecksum is true.
+   * If it is false, the resulting tree proto will have data checksums for each block, but an empty/unset data checksum
+   * for the container at the root of the tree.
+   */
+  public ContainerProtos.ContainerMerkleTree addDeletedBlocks(Collection<BlockData> blocks, boolean computeChecksum) {
+    for (BlockData block: blocks) {
+      long blockID = block.getLocalID();
+      BlockMerkleTreeWriter blockWriter = new BlockMerkleTreeWriter(blockID);
+      for (ContainerProtos.ChunkInfo chunkInfo: block.getChunks()) {
+        blockWriter.addChunks(new ChunkMerkleTreeWriter(chunkInfo, true));
+      }
+      blockWriter.markDeleted();
+      id2Block.put(blockID, blockWriter);
+    }
+    ContainerProtos.ContainerMerkleTree.Builder protoBuilder = toProtoBuilder();
+    if (!computeChecksum) {
+      protoBuilder.clearDataChecksum();
+    }
+    return protoBuilder.build();
+  }
+
   /**
    * Uses chunk hashes to compute all remaining hashes in the tree, and returns it as a protobuf object. No checksum
    * computation for the tree happens outside of this method.
    *
    * @return A complete protobuf object representation of this tree.
    */
-  public ContainerProtos.ContainerMerkleTree toProto() {
+  private ContainerProtos.ContainerMerkleTree.Builder toProtoBuilder() {
     // Compute checksums and return the result.
     ContainerProtos.ContainerMerkleTree.Builder containerTreeBuilder = ContainerProtos.ContainerMerkleTree.newBuilder();
     ChecksumByteBuffer checksumImpl = CHECKSUM_BUFFER_SUPPLIER.get();
@@ -133,8 +212,11 @@ public ContainerProtos.ContainerMerkleTree toProto() {
     checksumImpl.update(containerChecksumBuffer);
 
     return containerTreeBuilder
-        .setDataChecksum(checksumImpl.getValue())
-        .build();
+        .setDataChecksum(checksumImpl.getValue());
+  }
+
+  public ContainerProtos.ContainerMerkleTree toProto() {
+    return toProtoBuilder().build();
   }
 
   /**
@@ -145,10 +227,22 @@ private static class BlockMerkleTreeWriter {
     // Chunk order in the checksum is determined by their offset.
     private final SortedMap<Long, ChunkMerkleTreeWriter> offset2Chunk;
     private final long blockID;
+    private boolean deleted;
+    private Long dataChecksum;
 
     BlockMerkleTreeWriter(long blockID) {
       this.blockID = blockID;
       this.offset2Chunk = new TreeMap<>();
+      this.deleted = false;
+    }
+
+    public void markDeleted(long deletedDataChecksum) {
+      this.deleted = true;
+      this.dataChecksum = deletedDataChecksum;
+    }
+
+    public void markDeleted() {
+      this.deleted = true;
     }
 
     /**
@@ -163,6 +257,10 @@ public void addChunks(ChunkMerkleTreeWriter... chunks) {
       }
     }
 
+    public boolean isDeleted() {
+      return deleted;
+    }
+
     /**
      * Uses chunk hashes to compute a block hash for this tree, and returns it as a protobuf object. All block checksum
      * computation for the tree happens within this method.
@@ -171,6 +269,23 @@ public void addChunks(ChunkMerkleTreeWriter... chunks) {
      */
     public ContainerProtos.BlockMerkleTree toProto() {
       ContainerProtos.BlockMerkleTree.Builder blockTreeBuilder = ContainerProtos.BlockMerkleTree.newBuilder();
+      if (dataChecksum != null) {
+        blockTreeBuilder.setDataChecksum(dataChecksum);
+      } else {
+        setDataChecksumFromChunks(blockTreeBuilder);
+      }
+
+      if (deleted) {
+        blockTreeBuilder.clearChunkMerkleTree();
+      }
+
+      return blockTreeBuilder
+          .setBlockID(blockID)
+          .setDeleted(deleted)
+          .build();
+    }
+
+    private void setDataChecksumFromChunks(ContainerProtos.BlockMerkleTree.Builder blockTreeBuilder) {
       ChecksumByteBuffer checksumImpl = CHECKSUM_BUFFER_SUPPLIER.get();
       // Allocate space for block ID + all chunk checksums
       ByteBuffer blockChecksumBuffer = ByteBuffer.allocate(Long.BYTES * (1 + offset2Chunk.size()));
@@ -189,11 +304,7 @@ public ContainerProtos.BlockMerkleTree toProto() {
       }
       blockChecksumBuffer.flip();
       checksumImpl.update(blockChecksumBuffer);
-
-      return blockTreeBuilder
-          .setBlockID(blockID)
-          .setDataChecksum(checksumImpl.getValue())
-          .build();
+      blockTreeBuilder.setDataChecksum(checksumImpl.getValue());
     }
   }