HDDS-15931. Remove usage of internal netty metrics

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/NettyMetrics.java

@@ -17,15 +17,16 @@
 
 package org.apache.hadoop.hdds.utils;
 
-import static org.apache.ratis.thirdparty.io.netty.util.internal.PlatformDependent.maxDirectMemory;
-import static org.apache.ratis.thirdparty.io.netty.util.internal.PlatformDependent.usedDirectMemory;
-
+import java.lang.management.ManagementFactory;
+import java.util.List;
 import org.apache.hadoop.metrics2.MetricsCollector;
 import org.apache.hadoop.metrics2.MetricsInfo;
 import org.apache.hadoop.metrics2.MetricsRecordBuilder;
 import org.apache.hadoop.metrics2.MetricsSource;
 import org.apache.hadoop.metrics2.MetricsSystem;
 import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
+import org.apache.ratis.thirdparty.io.netty.buffer.ByteBufAllocator;
+import org.apache.ratis.thirdparty.io.netty.buffer.ByteBufAllocatorMetricProvider;
 
 /**
  * This class emits Netty metrics.
@@ -34,6 +35,13 @@ public final class NettyMetrics implements MetricsSource {
 
   public static final String SOURCE_NAME = NettyMetrics.class.getSimpleName();
 
+  private static final String MAX_DIRECT_MEMORY_FLAG =
+      "-XX:MaxDirectMemorySize=";
+
+  // JVM arguments are fixed for the process lifetime, so resolve the maximum
+  // direct memory once instead of on every getMetrics() call.
+  private static final long MAX_DIRECT_MEMORY = maxDirectMemory();
+
   public static NettyMetrics create() {
     MetricsSystem ms = DefaultMetricsSystem.instance();
     NettyMetrics metrics = new NettyMetrics();
@@ -46,14 +54,102 @@ public void getMetrics(MetricsCollector collector, boolean all) {
         .setContext("Netty metrics");
     recordBuilder
         .addGauge(MetricsInfos.USED_DIRECT_MEM, usedDirectMemory())
-        .addGauge(MetricsInfos.MAX_DIRECT_MEM, maxDirectMemory());
+        .addGauge(MetricsInfos.MAX_DIRECT_MEM, MAX_DIRECT_MEMORY);
   }
 
   public void unregister() {
     MetricsSystem ms = DefaultMetricsSystem.instance();
     ms.unregisterSource(SOURCE_NAME);
   }
 
+  /**
+   * Direct memory currently used by the default Netty {@link ByteBufAllocator}
+   * (the allocator Ratis and gRPC use). The previous implementation reported
+   * Netty's process-wide direct memory counter; this reports the default
+   * allocator's usage, which is close but not necessarily identical.
+   *
+   * @return used direct memory in bytes, or -1 if it cannot be determined
+   */
+  static long usedDirectMemory() {
+    ByteBufAllocator allocator = ByteBufAllocator.DEFAULT;
+    if (allocator instanceof ByteBufAllocatorMetricProvider) {
+      return ((ByteBufAllocatorMetricProvider) allocator).metric()
+          .usedDirectMemory();
+    }
+    return -1L;
+  }
+
+  /**
+   * Resolve the maximum direct memory using only public API, mirroring Netty's
+   * resolution order: the {@code io.netty.maxDirectMemory} system property, then
+   * the {@code -XX:MaxDirectMemorySize} JVM flag, then the maximum heap size.
+   * The JVM flag is read from the runtime input arguments (command line or
+   * {@code JAVA_TOOL_OPTIONS}); a value set by other means falls back to the
+   * maximum heap size.
+   *
+   * @return maximum direct memory in bytes
+   */
+  static long maxDirectMemory() {
+    return resolveMaxDirectMemory(
+        Long.getLong("io.netty.maxDirectMemory", -1L),
+        ManagementFactory.getRuntimeMXBean().getInputArguments(),
+        Runtime.getRuntime().maxMemory());
+  }
+
+  static long resolveMaxDirectMemory(long nettyProperty, List<String> jvmArgs,
+      long maxHeap) {
+    if (nettyProperty >= 0) {
+      return nettyProperty;
+    }
+    for (String arg : jvmArgs) {
+      if (arg.startsWith(MAX_DIRECT_MEMORY_FLAG)) {
+        long parsed =
+            parseSize(arg.substring(MAX_DIRECT_MEMORY_FLAG.length()));
+        if (parsed > 0) {
+          return parsed;
+        }
+      }
+    }
+    return maxHeap;
+  }
+
+  /**
+   * Parse a JVM memory size such as {@code "512m"}, {@code "1g"} or
+   * {@code "1073741824"} (1024-based units).
+   *
+   * @return the size in bytes, or -1 if the value cannot be parsed
+   */
+  static long parseSize(String value) {
+    String size = value.trim();
+    if (size.isEmpty()) {
+      return -1L;
+    }
+    long unit = 1L;
+    switch (Character.toLowerCase(size.charAt(size.length() - 1))) {
+    case 'k':
+      unit = 1L << 10;
+      break;
+    case 'm':
+      unit = 1L << 20;
+      break;
+    case 'g':
+      unit = 1L << 30;
+      break;
+    case 't':
+      unit = 1L << 40;
+      break;
+    default:
+      break;
+    }
+    String digits = unit == 1L ? size : size.substring(0, size.length() - 1);
+    try {
+      long number = Long.parseLong(digits.trim());
+      return number < 0 ? -1L : number * unit;
+    } catch (NumberFormatException e) {
+      return -1L;
+    }
+  }
+
   private enum MetricsInfos implements MetricsInfo {
     USED_DIRECT_MEM("Used direct memory."),
     MAX_DIRECT_MEM("Max direct memory.");

hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/utils/TestNettyMetrics.java

@@ -0,0 +1,89 @@
+/*
+ * 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.hdds.utils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.Arrays;
+import java.util.Collections;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests for {@link NettyMetrics}, in particular the public-API replacements for
+ * the direct memory gauges.
+ */
+class TestNettyMetrics {
+
+  private static final long MAX_HEAP = 8L << 30;
+
+  @Test
+  void resolveUsesNettyProperty() {
+    assertEquals(4096L, NettyMetrics.resolveMaxDirectMemory(
+        4096L, Collections.emptyList(), MAX_HEAP));
+    // The property takes precedence over the JVM flag, even when it is 0.
+    assertEquals(0L, NettyMetrics.resolveMaxDirectMemory(
+        0L, Arrays.asList("-XX:MaxDirectMemorySize=1g"), MAX_HEAP));
+  }
+
+  @Test
+  void resolveUsesMaxDirectMemorySizeFlag() {
+    assertEquals(512L << 20, NettyMetrics.resolveMaxDirectMemory(
+        -1L, Arrays.asList("-XX:MaxDirectMemorySize=512m"), MAX_HEAP));
+    assertEquals(1L << 30, NettyMetrics.resolveMaxDirectMemory(
+        -1L, Arrays.asList("-Xmx4g", "-XX:MaxDirectMemorySize=1g"), MAX_HEAP));
+    assertEquals(1073741824L, NettyMetrics.resolveMaxDirectMemory(
+        -1L, Arrays.asList("-XX:MaxDirectMemorySize=1073741824"), MAX_HEAP));
+  }
+
+  @Test
+  void resolveFallsBackToMaxHeap() {
+    assertEquals(MAX_HEAP, NettyMetrics.resolveMaxDirectMemory(
+        -1L, Collections.emptyList(), MAX_HEAP));
+    assertEquals(MAX_HEAP, NettyMetrics.resolveMaxDirectMemory(
+        -1L, Arrays.asList("-Xmx8g"), MAX_HEAP));
+    // 0 is not a positive limit, so fall back.
+    assertEquals(MAX_HEAP, NettyMetrics.resolveMaxDirectMemory(
+        -1L, Arrays.asList("-XX:MaxDirectMemorySize=0"), MAX_HEAP));
+    // Malformed value, so fall back.
+    assertEquals(MAX_HEAP, NettyMetrics.resolveMaxDirectMemory(
+        -1L, Arrays.asList("-XX:MaxDirectMemorySize=bogus"), MAX_HEAP));
+  }
+
+  @Test
+  void parseSizeVariants() {
+    assertEquals(1L << 10, NettyMetrics.parseSize("1k"));
+    assertEquals(1L << 20, NettyMetrics.parseSize("1m"));
+    assertEquals(1L << 30, NettyMetrics.parseSize("1G"));
+    assertEquals(1L << 40, NettyMetrics.parseSize("1t"));
+    assertEquals(2048L, NettyMetrics.parseSize("2048"));
+    assertEquals(-1L, NettyMetrics.parseSize(""));
+    assertEquals(-1L, NettyMetrics.parseSize("abc"));
+    assertEquals(-1L, NettyMetrics.parseSize("m"));
+  }
+
+  @Test
+  void usedDirectMemoryDoesNotThrow() {
+    assertThat(NettyMetrics.usedDirectMemory()).isGreaterThanOrEqualTo(-1L);
+  }
+
+  @Test
+  void maxDirectMemoryIsPositive() {
+    assertThat(NettyMetrics.maxDirectMemory()).isGreaterThan(0L);
+  }
+}