HDDS-10374. Make container scanner generate merkle trees during the scan

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

@@ -81,35 +81,45 @@ public void stop() {
    * The data merkle tree within the file is replaced with the {@code tree} parameter, but all other content of the
    * file remains unchanged.
    * Concurrent writes to the same file are coordinated internally.
+   * This method also updates the container's data checksum in the {@code data} parameter, which will be seen by SCM
+   * on container reports.
    */
   public ContainerProtos.ContainerChecksumInfo writeContainerDataTree(ContainerData data,
-                                                                      ContainerMerkleTreeWriter tree)
-      throws IOException {
+      ContainerMerkleTreeWriter tree) throws IOException {
     long containerID = data.getContainerID();
+    // If there is an error generating the tree and we cannot obtain a final checksum, use 0 to indicate a metadata
+    // failure.
+    long dataChecksum = 0;
+    ContainerProtos.ContainerChecksumInfo checksumInfo = null;
     Lock writeLock = getLock(containerID);
     writeLock.lock();
     try {
       ContainerProtos.ContainerChecksumInfo.Builder checksumInfoBuilder = null;
       try {
         // If the file is not present, we will create the data for the first time. This happens under a write lock.
-        checksumInfoBuilder = readBuilder(data)
-            .orElse(ContainerProtos.ContainerChecksumInfo.newBuilder());
+        checksumInfoBuilder = readBuilder(data).orElse(ContainerProtos.ContainerChecksumInfo.newBuilder());
       } catch (IOException ex) {
-        LOG.error("Failed to read container checksum tree file for container {}. Overwriting it with a new instance.",
+        LOG.error("Failed to read container checksum tree file for container {}. Creating a new instance.",
             containerID, ex);
         checksumInfoBuilder = ContainerProtos.ContainerChecksumInfo.newBuilder();
       }
 
-      ContainerProtos.ContainerChecksumInfo checksumInfo = checksumInfoBuilder
+      ContainerProtos.ContainerMerkleTree treeProto = captureLatencyNs(metrics.getCreateMerkleTreeLatencyNS(),
+          tree::toProto);
+      checksumInfoBuilder
           .setContainerID(containerID)
-          .setContainerMerkleTree(captureLatencyNs(metrics.getCreateMerkleTreeLatencyNS(), tree::toProto))
-          .build();
+          .setContainerMerkleTree(treeProto);
+      checksumInfo = checksumInfoBuilder.build();
       write(data, checksumInfo);
-      LOG.debug("Data merkle tree for container {} updated", containerID);
-      return checksumInfo;
+      // If write succeeds, update the checksum in memory. Otherwise 0 will be used to indicate the metadata failure.
+      dataChecksum = treeProto.getDataChecksum();
+      LOG.debug("Data merkle tree for container {} updated with container checksum {}", containerID, dataChecksum);
     } finally {
+      // Even if persisting the tree fails, we should still update the data checksum in memory to report back to SCM.
+      data.setDataChecksum(dataChecksum);
       writeLock.unlock();
     }
+    return checksumInfo;
   }
 
   /**
@@ -364,8 +374,6 @@ private void write(ContainerData data, ContainerProtos.ContainerChecksumInfo che
       throw new IOException("Error occurred when writing container merkle tree for containerID "
           + data.getContainerID(), ex);
     }
-    // Set in-memory data checksum.
-    data.setDataChecksum(checksumInfo.getContainerMerkleTree().getDataChecksum());
   }
 
   /**
@@ -388,7 +396,7 @@ public ContainerMerkleTreeMetrics getMetrics() {
     return this.metrics;
   }
 
-  public static boolean checksumFileExist(Container container) {
+  public static boolean checksumFileExist(Container<?> container) {
     File checksumFile = getContainerChecksumFile(container.getContainerData());
     return checksumFile.exists();
   }

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

@@ -60,10 +60,28 @@ public ContainerMerkleTreeWriter() {
    * If the block entry already exists, the chunks will be added to the existing chunks for that block.
    *
    * @param blockID The ID of the block that these chunks belong to.
+   * @param healthy True if there were no errors detected with these chunks. False indicates that all the chunks
+   *                being added had errors.
    * @param chunks A list of chunks to add to this block. The chunks will be sorted internally by their offset.
    */
-  public void addChunks(long blockID, Collection<ContainerProtos.ChunkInfo> chunks) {
-    id2Block.computeIfAbsent(blockID, BlockMerkleTreeWriter::new).addChunks(chunks);
+  public void addChunks(long blockID, boolean healthy, ContainerProtos.ChunkInfo... chunks) {
+    id2Block.computeIfAbsent(blockID, BlockMerkleTreeWriter::new).addChunks(healthy, chunks);
+  }
+
+  public void addChunks(long blockID, boolean healthy, Collection<ContainerProtos.ChunkInfo> chunks) {
+    for (ContainerProtos.ChunkInfo chunk: chunks) {
+      addChunks(blockID, healthy, chunk);
+    }
+  }
+
+  /**
+   * Adds an empty block to the tree. This method is not a pre-requisite to {@code addChunks}.
+   * If the block entry already exists, it will not be modified.
+   *
+   * @param blockID The ID of the empty block to add to the tree
+   */
+  public void addBlock(long blockID) {
+    addChunks(blockID, true);
   }
 
   /**
@@ -110,11 +128,13 @@ private static class BlockMerkleTreeWriter {
      * Adds the specified chunks to this block. The offset value of the chunk must be unique within the block,
      * otherwise it will overwrite the previous value at that offset.
      *
+     * @param healthy True if there were no errors detected with these chunks. False indicates that all the chunks
+     *                being added had errors.
      * @param chunks A list of chunks to add to this block.
      */
-    public void addChunks(Collection<ContainerProtos.ChunkInfo> chunks) {
+    public void addChunks(boolean healthy, ContainerProtos.ChunkInfo... chunks) {
       for (ContainerProtos.ChunkInfo chunk: chunks) {
-        offset2Chunk.put(chunk.getOffset(), new ChunkMerkleTreeWriter(chunk));
+        offset2Chunk.put(chunk.getOffset(), new ChunkMerkleTreeWriter(chunk, healthy));
       }
     }
 
@@ -160,10 +180,10 @@ private static class ChunkMerkleTreeWriter {
     private final boolean isHealthy;
     private final long dataChecksum;
 
-    ChunkMerkleTreeWriter(ContainerProtos.ChunkInfo chunk) {
+    ChunkMerkleTreeWriter(ContainerProtos.ChunkInfo chunk, boolean healthy) {
       length = chunk.getLen();
       offset = chunk.getOffset();
-      isHealthy = true;
+      isHealthy = healthy;
       ChecksumByteBuffer checksumImpl = CHECKSUM_BUFFER_SUPPLIER.get();
       for (ByteString checksum: chunk.getChecksumData().getChecksumsList()) {
         checksumImpl.update(checksum.asReadOnlyByteBuffer());

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainerCheck.java

@@ -188,7 +188,6 @@ public DataScanResult fullCheck(DataTransferThrottler throttler, Canceler cancel
 
     LOG.debug("Running data checks for container {}", containerID);
     try {
-      // TODO HDDS-10374 this tree will get updated with the container's contents as it is scanned.
       ContainerMerkleTreeWriter dataTree = new ContainerMerkleTreeWriter();
       List<ContainerScanError> dataErrors = scanData(dataTree, throttler, canceler);
       if (containerIsDeleted()) {
@@ -375,12 +374,15 @@ private List<ContainerScanError> scanBlock(DBHandle db, File dbFile, BlockData b
           // So, we need to make sure, chunk length > 0, before declaring
           // the missing chunk file.
           if (!block.getChunks().isEmpty() && block.getChunks().get(0).getLen() > 0) {
-            ContainerScanError error = new ContainerScanError(FailureType.MISSING_CHUNK_FILE,
+            ContainerScanError error = new ContainerScanError(FailureType.MISSING_DATA_FILE,
                 new File(containerDataFromDisk.getChunksPath()), new IOException("Missing chunk file " +
                 chunkFile.getAbsolutePath()));
             blockErrors.add(error);
           }
         } else if (chunk.getChecksumData().getType() != ContainerProtos.ChecksumType.NONE) {
+          // Before adding chunks, add a block entry to the tree to represent cases where the block exists but has no
+          // chunks.
+          currentTree.addBlock(block.getBlockID().getLocalID());
           int bytesPerChecksum = chunk.getChecksumData().getBytesPerChecksum();
           ByteBuffer buffer = BUFFER_POOL.getBuffer(bytesPerChecksum);
           // Keep scanning the block even if there are errors with individual chunks.
@@ -418,6 +420,14 @@ private static List<ContainerScanError> verifyChecksum(BlockData block,
 
     List<ContainerScanError> scanErrors = new ArrayList<>();
 
+    // Information used to populate the merkle tree. Chunk metadata will be the same, but we must fill in the
+    // checksums with what we actually observe.
+    ContainerProtos.ChunkInfo.Builder observedChunkBuilder = chunk.toBuilder();
+    ContainerProtos.ChecksumData.Builder observedChecksumData = chunk.getChecksumData().toBuilder();
+    observedChecksumData.clearChecksums();
+    boolean chunkHealthy = true;
+    boolean chunkMissing = false;
+
     ChecksumData checksumData =
         ChecksumData.getFromProtoBuf(chunk.getChecksumData());
     int checksumCount = checksumData.getChecksums().size();
@@ -430,10 +440,7 @@ private static List<ContainerScanError> verifyChecksum(BlockData block,
       if (layout == ContainerLayoutVersion.FILE_PER_BLOCK) {
         channel.position(chunk.getOffset());
       }
-      // Only report one error per chunk. Reporting corruption at every "bytes per checksum" interval will lead to a
-      // large amount of errors when a full chunk is corrupted.
-      boolean chunkHealthy = true;
-      for (int i = 0; i < checksumCount && chunkHealthy; i++) {
+      for (int i = 0; i < checksumCount; i++) {
         // limit last read for FILE_PER_BLOCK, to avoid reading next chunk
         if (layout == ContainerLayoutVersion.FILE_PER_BLOCK &&
             i == checksumCount - 1 &&
@@ -453,7 +460,11 @@ private static List<ContainerScanError> verifyChecksum(BlockData block,
         ByteString expected = checksumData.getChecksums().get(i);
         ByteString actual = cal.computeChecksum(buffer)
             .getChecksums().get(0);
-        if (!expected.equals(actual)) {
+        observedChecksumData.addChecksums(actual);
+        // Only report one error per chunk. Reporting corruption at every "bytes per checksum" interval will lead to a
+        // large amount of errors when a full chunk is corrupted.
+        // Continue scanning the chunk even after the first error so the full merkle tree can be built.
+        if (chunkHealthy && !expected.equals(actual)) {
           String message = String
               .format("Inconsistent read for chunk=%s" +
                   " checksum item %d" +
@@ -465,26 +476,46 @@ private static List<ContainerScanError> verifyChecksum(BlockData block,
                   StringUtils.bytes2Hex(expected.asReadOnlyByteBuffer()),
                   StringUtils.bytes2Hex(actual.asReadOnlyByteBuffer()),
                   block.getBlockID());
+          chunkHealthy = false;
           scanErrors.add(new ContainerScanError(FailureType.CORRUPT_CHUNK, chunkFile,
               new OzoneChecksumException(message)));
-          chunkHealthy = false;
         }
       }
-      // If all the checksums match, also check that the length stored in the metadata matches the number of bytes
-      // seen on the disk.
+
+      observedChunkBuilder.setLen(bytesRead);
+      // If we haven't seen any errors after scanning the whole chunk, verify that the length stored in the metadata
+      // matches the number of bytes seen on the disk.
       if (chunkHealthy && bytesRead != chunk.getLen()) {
-        String message = String
-            .format("Inconsistent read for chunk=%s expected length=%d"
-                    + " actual length=%d for block %s",
-                chunk.getChunkName(),
-                chunk.getLen(), bytesRead, block.getBlockID());
-        scanErrors.add(new ContainerScanError(FailureType.INCONSISTENT_CHUNK_LENGTH, chunkFile,
-            new IOException(message)));
+        if (bytesRead == 0) {
+          // If we could not find any data for the chunk, report it as missing.
+          chunkMissing = true;
+          chunkHealthy = false;
+          String message = String.format("Missing chunk=%s with expected length=%d for block %s",
+                  chunk.getChunkName(), chunk.getLen(), block.getBlockID());
+          scanErrors.add(new ContainerScanError(FailureType.MISSING_CHUNK, chunkFile, new IOException(message)));
+        } else {
+          // We found data for the chunk, but it was shorter than expected.
+          String message = String
+              .format("Inconsistent read for chunk=%s expected length=%d"
+                      + " actual length=%d for block %s",
+                  chunk.getChunkName(),
+                  chunk.getLen(), bytesRead, block.getBlockID());
+          chunkHealthy = false;
+          scanErrors.add(new ContainerScanError(FailureType.INCONSISTENT_CHUNK_LENGTH, chunkFile,
+              new IOException(message)));
+        }
       }
     } catch (IOException ex) {
-      scanErrors.add(new ContainerScanError(FailureType.MISSING_CHUNK_FILE, chunkFile, ex));
+      // An unknown error occurred trying to access the chunk. Report it as corrupted.
+      chunkHealthy = false;
+      scanErrors.add(new ContainerScanError(FailureType.CORRUPT_CHUNK, chunkFile, ex));
     }
 
+    // Missing chunks should not be added to the merkle tree.
+    if (!chunkMissing) {
+      observedChunkBuilder.setChecksumData(observedChecksumData);
+      currentTree.addChunks(block.getBlockID().getLocalID(), chunkHealthy, observedChunkBuilder.build());
+    }
     return scanErrors;
   }
 

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java

@@ -626,13 +626,18 @@ ContainerCommandResponseProto handleCloseContainer(
     return getSuccessResponse(request);
   }
 
-
   /**
-   * Create a Merkle tree for the container if it does not exist.
+   * Write the merkle tree for this container using the existing checksum metadata only. The data is not read or
+   * validated by this method, so it is expected to run quickly.
+   *
+   * If a checksum file already exists on the disk, this method will do nothing. The existing file would have either
+   * been made from the metadata or data itself so there is no need to recreate it from the metadata.
+   *
    * TODO: This method should be changed to private after HDDS-10374 is merged.
+   *
+   * @param container The container which will have a tree generated.
    */
-  @VisibleForTesting
-  public void createContainerMerkleTree(Container container) {
+  public void createContainerMerkleTreeFromMetadata(Container container) {
     if (ContainerChecksumTreeManager.checksumFileExist(container)) {
       return;
     }
@@ -1393,7 +1398,7 @@ public void markContainerForClose(Container container)
     } finally {
       container.writeUnlock();
     }
-    createContainerMerkleTree(container);
+    createContainerMerkleTreeFromMetadata(container);
     ContainerLogger.logClosing(container.getContainerData());
     sendICR(container);
   }
@@ -1426,7 +1431,6 @@ public void markContainerUnhealthy(Container container, ScanResult reason)
     } finally {
       container.writeUnlock();
     }
-    createContainerMerkleTree(container);
     // Even if the container file is corrupted/missing and the unhealthy
     // update fails, the unhealthy state is kept in memory and sent to
     // SCM. Write a corresponding entry to the container log as well.
@@ -1457,7 +1461,7 @@ public void quasiCloseContainer(Container container, String reason)
     } finally {
       container.writeUnlock();
     }
-    createContainerMerkleTree(container);
+    createContainerMerkleTreeFromMetadata(container);
     ContainerLogger.logQuasiClosed(container.getContainerData(), reason);
     sendICR(container);
   }
@@ -1491,7 +1495,7 @@ public void closeContainer(Container container)
     } finally {
       container.writeUnlock();
     }
-    createContainerMerkleTree(container);
+    createContainerMerkleTreeFromMetadata(container);
     ContainerLogger.logClosed(container.getContainerData());
     sendICR(container);
   }
@@ -1599,12 +1603,12 @@ private ContainerProtos.ContainerChecksumInfo updateAndGetContainerChecksum(KeyV
         BlockData blockData = blockIterator.nextBlock();
         List<ContainerProtos.ChunkInfo> chunkInfos = blockData.getChunks();
         // TODO: Add empty blocks to the merkle tree. Done in HDDS-10374, needs to be backported.
-        merkleTree.addChunks(blockData.getLocalID(), chunkInfos);
+        // Assume all chunks are healthy when building the tree from metadata. Scanner will identify corruption when
+        // it runs after.
+        merkleTree.addChunks(blockData.getLocalID(), true, chunkInfos);
       }
     }
-    ContainerProtos.ContainerChecksumInfo checksumInfo = checksumManager
-        .writeContainerDataTree(containerData, merkleTree);
-    return checksumInfo;
+    return checksumManager.writeContainerDataTree(containerData, merkleTree);
   }
 
   /**

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/BackgroundContainerDataScanner.java

@@ -92,6 +92,13 @@ public void scanContainer(Container<?> c)
     if (result.isDeleted()) {
       LOG.debug("Container [{}] has been deleted during the data scan.", containerId);
     } else {
+      // Merkle tree write failure should not abort the scanning process. Continue marking the scan as completed.
+      try {
+        checksumManager.writeContainerDataTree(containerData, result.getDataTree());
+      } catch (IOException ex) {
+        LOG.error("Failed to write container merkle tree for container {}", containerId, ex);
+      }
+
       if (!result.isHealthy()) {
         logUnhealthyScanResult(containerId, result, LOG);
 
@@ -103,17 +110,15 @@ public void scanContainer(Container<?> c)
           metrics.incNumUnHealthyContainers();
         }
       }
-      checksumManager.writeContainerDataTree(containerData, result.getDataTree());
       metrics.incNumContainersScanned();
     }
 
-    // Even if the container was deleted, mark the scan as completed since we already logged it as starting.
     Instant now = Instant.now();
-    logScanCompleted(containerData, now);
-
     if (!result.isDeleted()) {
       controller.updateDataScanTimestamp(containerId, now);
     }
+    // Even if the container was deleted, mark the scan as completed since we already logged it as starting.
+    logScanCompleted(containerData, now);
   }
 
   @Override

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/ContainerScanError.java

@@ -32,9 +32,10 @@ public enum FailureType {
     MISSING_METADATA_DIR,
     MISSING_CONTAINER_FILE,
     MISSING_CHUNKS_DIR,
-    MISSING_CHUNK_FILE,
+    MISSING_DATA_FILE,
     CORRUPT_CONTAINER_FILE,
     CORRUPT_CHUNK,
+    MISSING_CHUNK,
     INCONSISTENT_CHUNK_LENGTH,
     INACCESSIBLE_DB,
     WRITE_FAILURE,