HDDS-15960. Hadoop metrics leak detection

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

@@ -28,6 +28,8 @@
  */
 public final class ContainerCacheMetrics {
 
+  private static final String SOURCE_NAME = "ContainerCacheMetrics";
+
   @Metric("Rate to measure the db open latency")
   private MutableRate dbOpenLatency;
 
@@ -54,9 +56,7 @@ private ContainerCacheMetrics() {
 
   public static ContainerCacheMetrics create() {
     MetricsSystem ms = DefaultMetricsSystem.instance();
-    String name = "ContainerCacheMetrics";
-
-    return ms.register(name, "null", new ContainerCacheMetrics());
+    return ms.register(SOURCE_NAME, "null", new ContainerCacheMetrics());
   }
 
   public void incNumDbGetOps() {

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

@@ -360,6 +360,17 @@ public void shutdown() {
     }
     volumeMap.clear();
 
+    // Shut down failed volumes too: their VolumeInfoMetrics is registered on
+    // creation (HDDS-7086) and must be unregistered on shutdown.
+    for (StorageVolume volume : failedVolumeMap.values()) {
+      try {
+        volume.shutdown();
+      } catch (Exception ex) {
+        LOG.error("Failed to shutdown failed volume : " + volume.getStorageDir(), ex);
+      }
+    }
+    failedVolumeMap.clear();
+
     if (volumeHealthMetrics != null) {
       volumeHealthMetrics.unregister();
     }

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

@@ -25,6 +25,8 @@
 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.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.io.File;
@@ -39,6 +41,8 @@
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.hdds.scm.ScmConfigKeys;
 import org.apache.hadoop.metrics2.MetricsRecordBuilder;
+import org.apache.hadoop.metrics2.MetricsSystem;
+import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
 import org.apache.hadoop.ozone.OzoneConfigKeys;
 import org.apache.hadoop.ozone.container.common.utils.HddsVolumeUtil;
 import org.junit.jupiter.api.AfterEach;
@@ -191,7 +195,16 @@ void testFailVolumes(@TempDir File readOnlyVolumePath, @TempDir File volumePath)
     assertEquals(readOnlyVolumePath, volSet.getFailedVolumesList().get(0)
         .getStorageDir());
     assertNumVolumes(volSet, 1, 1);
+
+    // The failed volume's VolumeInfoMetrics is registered on creation
+    // (HDDS-7086) and must be unregistered when the volume set shuts down.
+    MetricsSystem ms = DefaultMetricsSystem.instance();
+    String failedVolumeMetrics = VolumeInfoMetrics.class.getSimpleName()
+        + "-" + readOnlyVolumePath.getAbsolutePath();
+    assertNotNull(ms.getSource(failedVolumeMetrics));
+
     volSet.shutdown();
+    assertNull(ms.getSource(failedVolumeMetrics));
   }
 
   @Test

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestSafeMode.java

@@ -59,7 +59,9 @@ class TestSafeMode {
   static void setup() {
     OzoneConfiguration conf = new OzoneConfiguration();
     clusterProvider = new MiniOzoneClusterProvider(
-        MiniOzoneCluster.newBuilder(conf), 2);
+        // Clusters overlap via MiniOzoneClusterProvider, so the metrics leak
+        // assertion would see a concurrent cluster's sources.
+        MiniOzoneCluster.newBuilder(conf).setMetricsLeakAssertEnabled(false), 2);
   }
 
   @BeforeEach

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/node/TestDecommissionAndMaintenance.java

@@ -148,7 +148,10 @@ public static void init() {
     conf.setFromObject(replicationConf);
 
     MiniOzoneCluster.Builder builder = MiniOzoneCluster.newBuilder(conf)
-        .setNumDatanodes(DATANODE_COUNT);
+        .setNumDatanodes(DATANODE_COUNT)
+        // Clusters overlap via MiniOzoneClusterProvider, so the metrics leak
+        // assertion would see a concurrent cluster's sources.
+        .setMetricsLeakAssertEnabled(false);
 
     clusterProvider = new MiniOzoneClusterProvider(builder, 9);
   }

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/upgrade/TestHDDSUpgrade.java

@@ -161,6 +161,9 @@ public static void initClass() {
     builder.setNumOfStorageContainerManagers(NUM_SCMS)
         .setSCMConfigurator(scmConfigurator)
         .setNumDatanodes(NUM_DATA_NODES)
+        // Clusters overlap via MiniOzoneClusterProvider, so the metrics leak
+        // assertion would see a concurrent cluster's sources.
+        .setMetricsLeakAssertEnabled(false)
         .setDatanodeFactory(UniformDatanodesFactory.newBuilder()
             .setLayoutVersion(HDDSLayoutFeature.INITIAL_VERSION.layoutVersion())
             .build());

hadoop-ozone/mini-cluster/src/main/java/org/apache/hadoop/ozone/MetricsLeakAssertion.java

@@ -0,0 +1,246 @@
+/*
+ * 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;
+
+import java.lang.reflect.Field;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
+import org.apache.hadoop.metrics2.MetricsSource;
+import org.apache.hadoop.metrics2.MetricsSystem;
+import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
+
+/**
+ * Asserts that no metrics sources remain registered in the
+ * {@link DefaultMetricsSystem} after a mini cluster is shut down.
+ *
+ * <p>Hadoop's {@code MetricsSystemImpl} does not remove registered sources on
+ * {@code stop()} or {@code shutdown()}; a metrics class that forgets to call
+ * {@code unregisterSource(...)} leaks its registration silently.  This helper
+ * inspects the private {@code allSources} map via reflection and fails the
+ * test if anything is still registered.
+ */
+public final class MetricsLeakAssertion {
+
+  private static final String ALL_SOURCES_FIELD = "allSources";
+
+  /**
+   * Metrics sources that are registered once per JVM (or once per service)
+   * and intentionally never unregistered, so they are expected to still be
+   * present after a mini cluster shuts down.  These are JVM-level singletons
+   * with no per-service owner; do not add per-service metrics here.
+   *
+   * <p>For the matching rules see {@link #isExpectedLeftover(String)}.
+   */
+  private static final List<String> EXPECTED_LEFTOVER_SOURCES = Arrays.asList(
+      "JvmMetrics*", // per service, via HddsServerUtil.initializeMetrics
+      "JvmMetricsCpu",
+      "UgiMetrics", // Hadoop security UGI metrics
+      "ManagedRocksObjectMetrics",
+      "ContainerCacheMetrics" // ContainerCache singleton, created once per JVM
+  );
+
+  /**
+   * Metrics sources that leak: they should be unregistered when the mini
+   * cluster shuts down but currently are not.  This is a transitional list to
+   * keep the integration suite green while the leaks are being fixed.  Each
+   * entry should be removed once the corresponding source unregisters
+   * properly; do not add new entries, fix the leak instead.
+   *
+   * <p>For the matching rules see {@link #isExpectedLeftover(String)}.
+   */
+  private static final List<String> TODO_LEFTOVER_SOURCES = Arrays.asList(
+      // RPC / HTTP server metrics (name embeds the listening port).
+      "RpcActivityForPort*",
+      "RpcDetailedActivityForPort*",
+      "HttpServer2*",
+      "LocalJobRunnerMetrics*",
+
+      // Datanode per-instance metrics.
+      "StorageContainerMetrics",
+      "ContainerDataScannerMetrics*", // name embeds the volume path
+      "ContainerMetadataScannerMetrics",
+      "On-demand container scanner metrics",
+      "BackgroundVolumeScannerMetrics",
+      "VolumeHealthMetrics-*",
+      "VolumeIOStats-*", // name embeds the volume path
+      // Name embeds the volume path.  Also intentionally kept registered for
+      // failed volumes (HDDS-7086), so only the healthy-volume case is a leak.
+      "VolumeInfoMetrics-*",
+      "CommandHandlerMetrics",
+      "HddsDispatcher",
+      "ECReconstructionMetrics",
+      "ReplicationSupervisorMetrics",
+      "ContainerReplicator/push",
+      "BlockDeletingService",
+      "GrpcMetrics",
+
+      // SCM metrics.
+      "SCMNodeMetrics",
+      "SCMContainerManagerMetrics",
+      "SCMContainerMetrics",
+      "SCMMetrics",
+      "SafeModeMetrics",
+      "ContainerBalancerMetrics",
+      "NodeDecommissionMetrics",
+      "SCMDatanodeProtocol",
+      "ScmBlockLocationProtocol",
+      "ScmContainerLocationProtocol",
+      "ScmSecurityProtocol",
+      "EventQueue*", // per event/handler pair, registered by SCM EventQueue
+
+      // OM metrics.
+      "OMMetrics",
+      "OMPerformanceMetrics",
+      "OMLockMetrics",
+      "OMHAMetrics",
+      "OMSnapshotDirectoryMetrics",
+      "OmSnapshotInternalMetrics",
+      "OmSnapshotMetrics",
+      "OmClientProtocol",
+      "DeletingServiceMetrics",
+      "KeyLifecycleServiceMetrics",
+      "BucketUtilizationMetrics",
+      "DelegationTokenSecretManagerMetrics",
+
+      // OM / SCM RocksDB table cache and DB metrics (name embeds path/uuid).
+      "Rocksdb_*",
+      "SSTFilePruningMetrics-*",
+      "DBCheckpointMetrics",
+      "*TableCache",
+
+      // Ratis / third-party infrastructure metrics.
+      "CSMMetricsgroup-*", // Ratis container state machine, embeds a random id
+      "CacheMetrics-*", // Hadoop cache metrics (XceiverClientManager, etc.)
+      "NettyMetrics*",
+      "ContainerClientMetrics1",
+      "ReconTaskMetrics",
+      "ReconTaskControllerMetrics"
+  );
+
+  private MetricsLeakAssertion() {
+  }
+
+  /**
+   * Throws an {@link AssertionError} if any metrics sources are still
+   * registered with the default metrics system, or if the expected
+   * {@code allSources} field cannot be found or read (e.g. a Hadoop version
+   * change restructured {@code MetricsSystemImpl}), so that a broken or
+   * missing check fails loudly instead of going unnoticed.
+   */
+  public static void assertNoLeaks() {
+    MetricsSystem ms = DefaultMetricsSystem.instance();
+    Field field = findAllSourcesField(ms.getClass());
+    if (field == null) {
+      throw new AssertionError("Cannot check for metrics leaks: '" + ALL_SOURCES_FIELD +
+          "' field not found on " + ms.getClass().getName() +
+          ". The metrics system implementation may have changed.");
+    }
+    final Map<String, MetricsSource> allSources;
+    try {
+      field.setAccessible(true);
+      Object value = field.get(ms);
+      if (!(value instanceof Map)) {
+        throw new AssertionError("Cannot check for metrics leaks: '" + ALL_SOURCES_FIELD +
+            "' on " + ms.getClass().getName() + " is not a Map.");
+      }
+      @SuppressWarnings("unchecked")
+      Map<String, MetricsSource> sources = (Map<String, MetricsSource>) value;
+      allSources = sources;
+    } catch (IllegalAccessException e) {
+      throw new AssertionError("Cannot check for metrics leaks: unable to access '" +
+          ALL_SOURCES_FIELD + "' on " + ms.getClass().getName() + ".", e);
+    }
+    Set<String> leaked = new TreeSet<>(allSources.keySet());
+    leaked.removeIf(MetricsLeakAssertion::isExpectedLeftover);
+    if (!leaked.isEmpty()) {
+      throw new AssertionError("Found " + leaked.size() +
+          " metrics source(s) still registered after cluster shutdown: " + leaked);
+    }
+  }
+
+  /**
+   * Matching rules for {@link #EXPECTED_LEFTOVER_SOURCES} and
+   * {@link #TODO_LEFTOVER_SOURCES}:
+   * <ul>
+   *   <li>An entry ending in {@code *} is a prefix match, used for sources
+   *   whose registered name embeds a random id, an absolute path, or a port
+   *   (e.g. {@code RpcActivityForPort15000}).</li>
+   *   <li>An entry starting with {@code *} is a suffix match, used for sources
+   *   whose registered name embeds a table-specific prefix (e.g.
+   *   {@code keyTableCache-1}).</li>
+   *   <li>Any other entry matches the source name exactly, or the name with a
+   *   numeric suffix the metrics system appends for repeated registrations
+   *   (e.g. {@code JvmMetrics-1}).</li>
+   * </ul>
+   */
+  private static boolean isExpectedLeftover(String name) {
+    return matches(EXPECTED_LEFTOVER_SOURCES, name) || matches(TODO_LEFTOVER_SOURCES, name);
+  }
+
+  private static boolean matches(List<String> entries, String name) {
+    for (String entry : entries) {
+      if (entry.startsWith("*")) {
+        // Suffix match, also tolerating a trailing numeric suffix (-N).
+        String suffix = entry.substring(1);
+        if (name.endsWith(suffix) || stripNumericSuffix(name).endsWith(suffix)) {
+          return true;
+        }
+      } else if (entry.endsWith("*")) {
+        if (name.startsWith(entry.substring(0, entry.length() - 1))) {
+          return true;
+        }
+      } else if (name.equals(entry) || name.startsWith(entry + "-")) {
+        return true;
+      }
+    }
+    return false;
+  }
+
+  private static String stripNumericSuffix(String name) {
+    int idx = name.lastIndexOf('-');
+    if (idx > 0 && idx < name.length() - 1) {
+      String tail = name.substring(idx + 1);
+      boolean numeric = true;
+      for (int i = 0; i < tail.length(); i++) {
+        if (!Character.isDigit(tail.charAt(i))) {
+          numeric = false;
+          break;
+        }
+      }
+      if (numeric) {
+        return name.substring(0, idx);
+      }
+    }
+    return name;
+  }
+
+  private static Field findAllSourcesField(Class<?> clazz) {
+    Class<?> current = clazz;
+    while (current != null) {
+      try {
+        return current.getDeclaredField(ALL_SOURCES_FIELD);
+      } catch (NoSuchFieldException e) {
+        current = current.getSuperclass();
+      }
+    }
+    return null;
+  }
+}

hadoop-ozone/mini-cluster/src/main/java/org/apache/hadoop/ozone/MiniOzoneCluster.java

@@ -262,6 +262,7 @@ abstract class Builder {
 
     protected int numOfDatanodes = 3;
     protected boolean  startDataNodes = true;
+    protected boolean metricsLeakAssertEnabled = true;
     protected CertificateClient certClient;
     protected SecretKeyClient secretKeyClient;
     protected DatanodeFactory dnFactory = UniformDatanodesFactory.newBuilder().build();
@@ -354,6 +355,17 @@ public Builder setStartDataNodes(boolean nodes) {
       return this;
     }
 
+    /**
+     * Whether to assert that no metrics sources leak on cluster shutdown.
+     * Disable for tests that run multiple clusters concurrently (e.g. via
+     * MiniOzoneClusterProvider), because the metrics registry is shared
+     * JVM-wide and a concurrent cluster's sources would be flagged as leaks.
+     */
+    public Builder setMetricsLeakAssertEnabled(boolean enabled) {
+      this.metricsLeakAssertEnabled = enabled;
+      return this;
+    }
+
     public Builder setCertificateClient(CertificateClient client) {
       this.certClient = client;
       return this;