HDDS-15455. Implement Custom DataNode Container Directory Discovery and Duplicate Detection (#10414).

Co-authored-by: Sreeja Chintalapati <schintalapati@Sreejas-MacBook-Pro.local>

Author: sreejasahithi

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/utils/StorageVolumeUtil.java

@@ -28,6 +28,7 @@
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.hadoop.ozone.common.InconsistentStorageStateException;
+import org.apache.hadoop.ozone.common.Storage;
 import org.apache.hadoop.ozone.container.common.HDDSVolumeLayoutVersion;
 import org.apache.hadoop.ozone.container.common.volume.DbVolume;
 import org.apache.hadoop.ozone.container.common.volume.HddsVolume;
@@ -274,4 +275,31 @@ public static boolean checkVolume(StorageVolume volume, String scmId,
 
     return success;
   }
+  
+  public static File resolveContainerCurrentDir(
+      File hddsRoot, String clusterId, File[] storageDirs)
+      throws InconsistentStorageStateException {
+
+    File clusterIdDir = new File(hddsRoot, clusterId);
+    //The subdirectory we should verify containers within.
+    // If this volume was formatted pre SCM HA, this will be the SCM ID.
+    // A cluster ID symlink will exist in this case only if this cluster is
+    // finalized for SCM HA.
+    // If the volume was formatted post SCM HA, this will be the cluster ID.
+    File idDir = clusterIdDir;
+
+    if (storageDirs.length == 1 && !clusterIdDir.exists()) {
+      // If the one directory is not the cluster ID directory, assume it is
+      // the old SCM ID directory used before SCM HA.
+      idDir = storageDirs[0];
+    } else if (!clusterIdDir.exists()) {
+      // There are 1 or more storage directories. We only care about the
+      // cluster ID directory.
+      throw new InconsistentStorageStateException(
+          "Volume " + hddsRoot + " is in an inconsistent state. Expected cluster ID directory "
+              + clusterIdDir + " not found.");
+    }
+
+    return new File(idDir, Storage.STORAGE_DIR_CURRENT);
+  }
 }

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

@@ -28,11 +28,12 @@
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
 import org.apache.hadoop.hdds.scm.container.ContainerID;
 import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
-import org.apache.hadoop.ozone.common.Storage;
+import org.apache.hadoop.ozone.common.InconsistentStorageStateException;
 import org.apache.hadoop.ozone.container.common.helpers.ContainerUtils;
 import org.apache.hadoop.ozone.container.common.impl.ContainerData;
 import org.apache.hadoop.ozone.container.common.impl.ContainerDataYaml;
 import org.apache.hadoop.ozone.container.common.impl.ContainerSet;
+import org.apache.hadoop.ozone.container.common.utils.StorageVolumeUtil;
 import org.apache.hadoop.ozone.container.common.volume.HddsVolume;
 import org.apache.hadoop.ozone.container.common.volume.MutableVolumeSet;
 import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainer;
@@ -119,32 +120,19 @@ public void readVolume(File hddsVolumeRootDir) {
     // by HddsUtil#checkVolume once we have a cluster ID from SCM. No
     // operations to perform here in that case.
     if (storageDirs.length > 0) {
-      File clusterIDDir = new File(hddsVolumeRootDir,
-          hddsVolume.getClusterID());
-      // The subdirectory we should verify containers within.
-      // If this volume was formatted pre SCM HA, this will be the SCM ID.
-      // A cluster ID symlink will exist in this case only if this cluster is
-      // finalized for SCM HA.
-      // If the volume was formatted post SCM HA, this will be the cluster ID.
-      File idDir = clusterIDDir;
-      if (storageDirs.length == 1 && !clusterIDDir.exists()) {
-        // If the one directory is not the cluster ID directory, assume it is
-        // the old SCM ID directory used before SCM HA.
-        idDir = storageDirs[0];
-      } else {
-        // There are 1 or more storage directories. We only care about the
-        // cluster ID directory.
-        if (!clusterIDDir.exists()) {
-          LOG.error("Volume {} is in an inconsistent state. Expected " +
-              "clusterID directory {} not found.", hddsVolumeRootDir,
-              clusterIDDir);
-          volumeSet.failVolume(hddsVolumeRootDir.getPath());
-          return;
-        }
+      File currentDir;
+      try {
+        currentDir = StorageVolumeUtil.resolveContainerCurrentDir(hddsVolumeRootDir, 
+            hddsVolume.getClusterID(), storageDirs);
+      } catch (InconsistentStorageStateException e) {
+        LOG.error("Volume {} is in an inconsistent state. Expected " +
+                "clusterID directory {} not found.", hddsVolumeRootDir,
+            new File(hddsVolumeRootDir, hddsVolume.getClusterID()));
+        volumeSet.failVolume(hddsVolumeRootDir.getPath());
+        return;
       }
 
       LOG.info("Start to verify containers on volume {}", hddsVolumeRootDir);
-      File currentDir = new File(idDir, Storage.STORAGE_DIR_CURRENT);
       File[] containerTopDirs = currentDir.listFiles();
       if (containerTopDirs != null && containerTopDirs.length > 0) {
         for (File containerTopDir : containerTopDirs) {

hadoop-ozone/cli-debug/src/main/java/org/apache/hadoop/ozone/debug/datanode/container/ContainerCommands.java

@@ -58,6 +58,7 @@
 import org.apache.hadoop.ozone.container.ozoneimpl.ContainerController;
 import org.apache.hadoop.ozone.container.ozoneimpl.ContainerReader;
 import org.apache.hadoop.ozone.container.upgrade.VersionedDatanodeFeatures;
+import org.apache.hadoop.ozone.debug.datanode.container.analyze.AnalyzeSubcommand;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import picocli.CommandLine.Command;
@@ -75,7 +76,8 @@
         ListSubcommand.class,
         InfoSubcommand.class,
         ExportSubcommand.class,
-        InspectSubcommand.class
+        InspectSubcommand.class,
+        AnalyzeSubcommand.class
     })
 public class ContainerCommands extends AbstractSubcommand {
 

hadoop-ozone/cli-debug/src/main/java/org/apache/hadoop/ozone/debug/datanode/container/analyze/AnalyzeSubcommand.java

@@ -0,0 +1,107 @@
+/*
+ * 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.debug.datanode.container.analyze;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import org.apache.hadoop.hdds.cli.AbstractSubcommand;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import picocli.CommandLine;
+import picocli.CommandLine.Command;
+
+/**
+ * {@code ozone debug datanode container analyze}.
+ *
+ * <p>Compares on-disk container directories on this DataNode against SCM
+ * metadata to report inconsistencies.
+ */
+@Command(
+    name = "analyze",
+    description = "Analyze container consistency between on-disk container " +
+            "directories on this DataNode and SCM metadata. Must be run locally on a DataNode.")
+public class AnalyzeSubcommand extends AbstractSubcommand implements Callable<Void> {
+  @CommandLine.Option(names = {"--count"},
+          defaultValue = "20",
+          description = "Number of containers to display")
+  private int count;
+
+  @Override
+  public Void call() throws Exception {
+    if (count < 1) {
+      throw new IOException("Count must be an integer greater than 0.");
+    }
+    OzoneConfiguration conf = getOzoneConf();
+    ContainerScanResult scanResult = ContainerDirectoryScanner.scan(conf);
+    Map<Long, List<ContainerDiskOccurrence>> enrichedDuplicates =
+        ContainerDirectoryScanner.enrichDuplicates(scanResult.getDuplicates());
+
+    // TODO: SCM metadata lookup from --scm-db when provided.
+    // TODO: For each id in scanResult.getSingles().keySet() classified NOT_IN_SCM or DELETED:
+    //   enrichOccurrence(id, scanResult.getSingles().get(id)) and report.
+    // TODO: For each id in enrichedDuplicates.keySet() classified NOT_IN_SCM or DELETED:
+    //   enrichedDuplicates.get(id) is already enriched — just report.
+
+    printDuplicates(enrichedDuplicates);
+    printVolumeScanErrors(scanResult.getVolumeScanErrors());
+    return null;
+  }
+
+  private void printDuplicates(Map<Long, List<ContainerDiskOccurrence>> duplicates) {
+    long totalDuplicateIds = duplicates.size();
+    out().printf("Number of containers with duplicate container directories on this DataNode: %d%n", totalDuplicateIds);
+
+    if (totalDuplicateIds == 0) {
+      return;
+    }
+
+    if (totalDuplicateIds > count) {
+      out().printf("Showing first %d:%n", count);
+    }
+
+    duplicates.entrySet().stream()
+        .sorted(Map.Entry.comparingByKey())
+        .limit(count)
+        .forEach(entry -> {
+          long containerId = entry.getKey();
+          List<ContainerDiskOccurrence> occurrences = entry.getValue();
+          out().printf("Container %d (%d occurrences):%n", containerId, occurrences.size());
+          for (ContainerDiskOccurrence o : occurrences) {
+            out().printf("  path=%s%n", o.getContainerPath());
+            if (o.isSizeKnown()) {
+              out().printf("  status=%s size=%d bytes%n", o.getStatus(), o.getSizeBytes());
+            } else {
+              out().printf("  status=%s size=unavailable (failed to compute directory size)%n",
+                  o.getStatus());
+            }
+            out().println();
+          }
+        });
+  }
+
+  private void printVolumeScanErrors(List<String> volumeScanErrors) {
+    if (volumeScanErrors.isEmpty()) {
+      return;
+    }
+    err().printf("%nVolumes that failed to scan (%d):%n", volumeScanErrors.size());
+    for (String error : volumeScanErrors) {
+      err().printf("  %s%n", error);
+    }
+  }
+}