HDDS-16119. Fix DatanodeStorageMetrics metrics-system deadlock and mini-cluster source leak

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/volume/DatanodeStorageMetrics.java

@@ -30,10 +30,13 @@
 
 /**
  * Node-level storage totals for a DataNode, aggregated over its HDDS data volumes only
- * ({@code VolumeType.DATA_VOLUME}) via {@link MutableVolumeSet#getStorageReport()}.
+ * ({@code VolumeType.DATA_VOLUME}) via {@link MutableVolumeSet#getStorageReportSnapshot()}.
  * This is the same scope as the {@code storageReport} entries produced by
  * {@code OzoneContainer.getNodeReport()}; meta and DB volumes are excluded.
  * Registered as {@code Hadoop:service=HddsDatanode,name=DatanodeStorageMetrics}.
+ * In mini-cluster mode, where many datanodes share one metrics system, the name
+ * is suffixed with the datanode UUID ({@code DatanodeStorageMetrics-<uuid>}) so
+ * registration and unregistration stay unique per datanode.
  */
 @Metrics(about = "Ozone DataNode node-level storage totals",
     context = OzoneConsts.OZONE)
@@ -53,28 +56,41 @@ public final class DatanodeStorageMetrics implements MetricsSource {
 
   private final MetricsRegistry registry;
   private final MutableVolumeSet volumeSet;
+  private final String sourceName;
 
-  private DatanodeStorageMetrics(MutableVolumeSet volumeSet) {
+  private DatanodeStorageMetrics(MutableVolumeSet volumeSet, String sourceName) {
     this.volumeSet = volumeSet;
-    this.registry = new MetricsRegistry(SOURCE_NAME);
+    this.sourceName = sourceName;
+    this.registry = new MetricsRegistry(sourceName);
   }
 
   /**
    * Creates a new {@code DatanodeStorageMetrics} instance and registers it
    * with the default Metrics2 system.
    */
   public static DatanodeStorageMetrics create(MutableVolumeSet volumeSet) {
-    DatanodeStorageMetrics datanodeStorageMetrics = new DatanodeStorageMetrics(volumeSet);
-    DefaultMetricsSystem.instance().register(
-        SOURCE_NAME, "DataNode node-level storage totals", datanodeStorageMetrics);
+    // In mini-cluster mode many datanodes share one metrics system, so the
+    // metrics system uniquifies the constant source name on registration
+    // (DatanodeStorageMetrics-1, -2, ...). Unregistering by the constant base
+    // name would then leak every source past the first, and each leaked source
+    // pins a shut-down datanode's MutableVolumeSet. Make the name unique per
+    // datanode up front so register and unregister stay symmetric. In
+    // production there is one instance per JVM, so keep the plain name for
+    // stable JMX and Prometheus metric names.
+    String sourceName = DefaultMetricsSystem.inMiniClusterMode()
+        ? SOURCE_NAME + '-' + volumeSet.getDatanodeUuid()
+        : SOURCE_NAME;
+    DatanodeStorageMetrics datanodeStorageMetrics = new DatanodeStorageMetrics(volumeSet, sourceName);
+    DefaultMetricsSystem.instance().register(sourceName,
+        "DataNode node-level storage totals", datanodeStorageMetrics);
     return datanodeStorageMetrics;
   }
 
   /**
    * Unregisters this source from the Metrics2 system.
    */
   public void unregister() {
-    DefaultMetricsSystem.instance().unregisterSource(SOURCE_NAME);
+    DefaultMetricsSystem.instance().unregisterSource(sourceName);
   }
 
   /**
@@ -83,12 +99,17 @@ public void unregister() {
    */
   @Override
   public void getMetrics(MetricsCollector collector, boolean all) {
-    MetricsRecordBuilder builder = collector.addRecord(SOURCE_NAME);
+    MetricsRecordBuilder builder = collector.addRecord(sourceName);
     registry.snapshot(builder, all);
 
     long capacity = 0L;
     long used = 0L;
-    for (StorageLocationReport report : volumeSet.getStorageReport()) {
+    // getMetrics() runs while the DefaultMetricsSystem monitor is held. Read a
+    // lock-free snapshot instead of getStorageReport(), which takes the
+    // volume-set lock: a volume-failure handler holds that lock while
+    // unregistering volume metrics (which needs the same monitor), so locking
+    // here can deadlock the metrics system.
+    for (StorageLocationReport report : volumeSet.getStorageReportSnapshot()) {
       capacity = Math.addExact(capacity, report.getCapacity());
       used = Math.addExact(used, report.getScmUsed());
     }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/volume/MutableVolumeSet.java

@@ -28,6 +28,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.function.Function;
+import java.util.stream.Stream;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
 import org.apache.hadoop.hdds.fs.SpaceUsageCheckFactory;
 import org.apache.hadoop.hdds.scm.ScmConfigKeys;
@@ -404,6 +405,16 @@ public boolean hasEnoughVolumes() {
     return hasEnoughVolumes;
   }
 
+  /**
+   * Returns a consistent snapshot of the storage reports under the volume-set
+   * read lock.
+   *
+   * <p>Do not call this from metrics collection (or any caller that may hold the
+   * {@code DefaultMetricsSystem} monitor): a volume-failure handler holds the
+   * volume-set write lock while unregistering volume metrics under that same
+   * monitor, so taking the read lock here can deadlock the metrics system. Use
+   * {@link #getStorageReportSnapshot()} from those paths instead.
+   */
   public StorageLocationReport[] getStorageReport() {
     this.readLock();
     try {
@@ -421,6 +432,32 @@ public StorageLocationReport[] getStorageReport() {
     }
   }
 
+  /**
+   * Lock-free variant of {@link #getStorageReport()}. Both {@code volumeMap}
+   * and {@code failedVolumeMap} are {@link ConcurrentHashMap}s, so this returns
+   * a weakly-consistent snapshot (mirroring {@link #getVolumesList()}) without
+   * acquiring the volume-set lock.
+   *
+   * <p>Use this from callers that must not block on the volume-set lock. In
+   * particular, metrics sampling runs while the {@code DefaultMetricsSystem}
+   * monitor is held, and a volume-failure handler holds the volume-set write
+   * lock while unregistering volume metrics (which needs that same monitor);
+   * acquiring the volume-set lock from the sampling thread can therefore
+   * deadlock the whole metrics system.
+   */
+  public StorageLocationReport[] getStorageReportSnapshot() {
+    // volumeMap and failedVolumeMap are ConcurrentHashMaps; their value streams
+    // are weakly consistent, so no lock is needed here (same guarantee as
+    // getVolumesList()).
+    return Stream.concat(volumeMap.values().stream(), failedVolumeMap.values().stream())
+        .map(StorageVolume::getReport)
+        .toArray(StorageLocationReport[]::new);
+  }
+
+  public String getDatanodeUuid() {
+    return datanodeUuid;
+  }
+
   public StorageVolume.VolumeType getVolumeType() {
     return volumeType;
   }

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestDatanodeStorageMetrics.java

@@ -21,9 +21,12 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+import java.util.UUID;
 import org.apache.hadoop.metrics2.AbstractMetric;
+import org.apache.hadoop.metrics2.MetricsSystem;
 import org.apache.hadoop.metrics2.impl.MetricsCollectorImpl;
 import org.apache.hadoop.metrics2.impl.MetricsRecordImpl;
+import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
 import org.apache.hadoop.ozone.container.common.impl.StorageLocationReport;
 import org.junit.jupiter.api.Test;
 
@@ -51,7 +54,7 @@ void testAggregationAcrossTwoVolumes() {
         .build();
 
     MutableVolumeSet volumeSet = mock(MutableVolumeSet.class);
-    when(volumeSet.getStorageReport())
+    when(volumeSet.getStorageReportSnapshot())
         .thenReturn(new StorageLocationReport[]{vol1, vol2});
 
     DatanodeStorageMetrics metrics = DatanodeStorageMetrics.create(volumeSet);
@@ -78,7 +81,7 @@ void testAggregationAcrossTwoVolumes() {
   void testZeroCapacityReturnsZeroPercentage() {
     // No volumes → capacity=0, used=0; OzoneUsedPercentage must be 0.0, not NaN.
     MutableVolumeSet volumeSet = mock(MutableVolumeSet.class);
-    when(volumeSet.getStorageReport()).thenReturn(new StorageLocationReport[0]);
+    when(volumeSet.getStorageReportSnapshot()).thenReturn(new StorageLocationReport[0]);
 
     DatanodeStorageMetrics metrics = DatanodeStorageMetrics.create(volumeSet);
     try {
@@ -94,6 +97,51 @@ void testZeroCapacityReturnsZeroPercentage() {
     }
   }
 
+  @Test
+  void testNoSourceLeakInMiniClusterMode() {
+    // In mini-cluster mode many datanodes share one metrics system. Register
+    // and unregister must be symmetric so no source (and its pinned volume set)
+    // leaks. The pre-fix code registered a constant name (uniquified to -N) but
+    // unregistered the base name, leaking every datanode past the first.
+    boolean prev = DefaultMetricsSystem.inMiniClusterMode();
+    DefaultMetricsSystem.setMiniClusterMode(true);
+    MetricsSystem ms = DefaultMetricsSystem.instance();
+    String uuidA = "dn-" + UUID.randomUUID();
+    String uuidB = "dn-" + UUID.randomUUID();
+    String nameA = DatanodeStorageMetrics.SOURCE_NAME + '-' + uuidA;
+    String nameB = DatanodeStorageMetrics.SOURCE_NAME + '-' + uuidB;
+    DatanodeStorageMetrics a = null;
+    DatanodeStorageMetrics b = null;
+    try {
+      a = DatanodeStorageMetrics.create(mockVolumeSet(uuidA));
+      b = DatanodeStorageMetrics.create(mockVolumeSet(uuidB));
+      assertThat(ms.getSource(nameA)).isNotNull();
+      assertThat(ms.getSource(nameB)).isNotNull();
+
+      a.unregister();
+      b.unregister();
+      assertThat(ms.getSource(nameA)).isNull();
+      assertThat(ms.getSource(nameB)).isNull();
+    } finally {
+      // Do not leak sources into other tests if an assertion above fails.
+      // unregister() is idempotent, so a repeat after the happy path is a no-op.
+      if (a != null) {
+        a.unregister();
+      }
+      if (b != null) {
+        b.unregister();
+      }
+      DefaultMetricsSystem.setMiniClusterMode(prev);
+    }
+  }
+
+  private static MutableVolumeSet mockVolumeSet(String datanodeUuid) {
+    MutableVolumeSet volumeSet = mock(MutableVolumeSet.class);
+    when(volumeSet.getDatanodeUuid()).thenReturn(datanodeUuid);
+    when(volumeSet.getStorageReportSnapshot()).thenReturn(new StorageLocationReport[0]);
+    return volumeSet;
+  }
+
   private static long findLong(Iterable<AbstractMetric> metrics, String name) {
     for (AbstractMetric m : metrics) {
       if (name.equals(m.name())) {

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestVolumeSet.java

@@ -25,6 +25,7 @@
 import static org.assertj.core.api.Assumptions.assumeThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.io.File;
@@ -34,6 +35,9 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import org.apache.commons.io.FileUtils;
 import org.apache.hadoop.hdds.HddsConfigKeys;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
@@ -194,6 +198,29 @@ void testFailVolumes(@TempDir File readOnlyVolumePath, @TempDir File volumePath)
     volSet.shutdown();
   }
 
+  @Test
+  public void testStorageReportSnapshotDoesNotBlockOnWriteLock() throws Exception {
+    // Regression test: DatanodeStorageMetrics.getMetrics() samples the volume
+    // set while the DefaultMetricsSystem monitor is held. It must not block on
+    // the volume-set lock, otherwise it deadlocks with a volume-failure handler
+    // that holds the write lock while unregistering volume metrics.
+    volumeSet.writeLock();
+    try {
+      CompletableFuture<Integer> snapshot = CompletableFuture.supplyAsync(
+          () -> volumeSet.getStorageReportSnapshot().length);
+      // The snapshot must return without waiting for the write lock.
+      assertEquals(2, snapshot.get(10, TimeUnit.SECONDS));
+
+      // Sanity check that the lock is genuinely held: the locking variant does
+      // block behind the write lock (it must time out here).
+      CompletableFuture<Integer> locked = CompletableFuture.supplyAsync(
+          () -> volumeSet.getStorageReport().length);
+      assertThrows(TimeoutException.class, () -> locked.get(2, TimeUnit.SECONDS));
+    } finally {
+      volumeSet.writeUnlock();
+    }
+  }
+
   @Test
   public void testInterrupt() throws Exception {
     Method method = this.volumeSet.getClass()

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/dn/TestDatanodeStorageMetricsIntegration.java

@@ -123,7 +123,12 @@ void storageMetricsReflectWrittenData() throws Exception {
    * Each call re-reads the underlying storage reports — do not mix values
    * from different calls when checking invariants across gauges.
    */
-  private static MetricsRecordBuilder storageMetrics() {
-    return getMetrics(DatanodeStorageMetrics.SOURCE_NAME);
+  private MetricsRecordBuilder storageMetrics() {
+    // In mini-cluster mode the source name is made unique per datanode (to keep
+    // metrics registration and unregistration symmetric and avoid the
+    // shared-JVM source leak), so look it up by the per-datanode name.
+    MutableVolumeSet volumeSet = cluster.getHddsDatanodes().get(0)
+        .getDatanodeStateMachine().getContainer().getVolumeSet();
+    return getMetrics(DatanodeStorageMetrics.SOURCE_NAME + '-' + volumeSet.getDatanodeUuid());
   }
 }