CASSANDRA-21556: Add metrics about Netty memory usage

src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java

@@ -144,6 +144,7 @@ public class CassandraMetricsRegistry extends MetricRegistry
                                    .add(MemtablePool.TYPE_NAME)
                                    .add(MessagingMetrics.TYPE_NAME)
                                    .add(MutualTlsMetrics.TYPE_NAME)
+                                   .add(NettyMemoryMetrics.TYPE_NAME)
                                    .add(PaxosMetrics.TYPE_NAME)
                                    .add(ReadRepairMetrics.TYPE_NAME)
                                    .add(RepairMetrics.TYPE_NAME)

src/java/org/apache/cassandra/metrics/NettyMemoryMetrics.java

@@ -0,0 +1,79 @@
+/*
+ * 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.cassandra.metrics;
+
+import com.codahale.metrics.Gauge;
+
+import io.netty.util.internal.PlatformDependent;
+
+import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics;
+
+/**
+ * Exposes the direct (off-heap) memory that Netty accounts for internally, process-wide.
+ *
+ * <p>Netty allocates direct memory through {@code Unsafe.allocateMemory} (the "no cleaner" path, enabled in Cassandra
+ * by {@code -Dio.netty.tryReflectionSetAccessible=true}), which bypasses {@code java.nio.Bits} entirely. Netty
+ * therefore maintains its own counter and its own limit, independent of the JDK's direct memory accounting. Both
+ * budgets default to {@code -XX:MaxDirectMemorySize}, so the two can each be filled independently and the process can
+ * hold considerably more direct memory than that setting alone suggests.
+ *
+ * <p>Consequences for interpreting these metrics:
+ * <ul>
+ *     <li>{@link #USED_DIRECT_MEMORY} aggregates across every Netty allocator in the process, but only covers memory
+ *     Netty itself allocated. It does <em>not</em> include Cassandra's {@code BufferPool}, which calls
+ *     {@code ByteBuffer.allocateDirect} directly and is reported separately by {@link BufferPoolMetrics}. Internode
+ *     messaging, streaming and native protocol v5+ channel buffers all go through {@code BufferPool}, not here.</li>
+ *     <li>The JDK-side counterpart ({@code java.nio.Bits}) is reported by {@code nodetool gcstats} as the allocated /
+ *     max / reserved direct memory values. That bucket and this one are disjoint; a complete picture requires both.</li>
+ * </ul>
+ *
+ * <p>Both gauges report {@code -1} when Netty's counter is unavailable, which happens when the no-cleaner path is
+ * disabled.
+ */
+public final class NettyMemoryMetrics
+{
+    public static final String TYPE_NAME = "NettyMemory";
+
+    /**
+     * Direct memory currently allocated by Netty and not yet freed, in bytes. Reflects what Netty has reserved from
+     * the OS rather than what is actively in use, since pooled arenas retain chunks for reuse.
+     */
+    public static final String USED_DIRECT_MEMORY = "UsedDirectMemory";
+
+    /**
+     * The direct memory ceiling Netty enforces against {@link #USED_DIRECT_MEMORY}, in bytes. Derived from
+     * {@code -Dio.netty.maxDirectMemory} when set, otherwise from {@code -XX:MaxDirectMemorySize}. Exceeding it raises
+     * Netty's {@code OutOfDirectMemoryError}, which is distinct from the JDK's
+     * {@code OutOfMemoryError: Direct buffer memory}.
+     */
+    public static final String DIRECT_MEMORY_LIMIT = "DirectMemoryLimit";
+
+    private NettyMemoryMetrics()
+    {
+    }
+
+    public static void register()
+    {
+        MetricNameFactory factory = new DefaultNameFactory(TYPE_NAME);
+
+        Metrics.register(factory.createMetricName(USED_DIRECT_MEMORY),
+                         (Gauge<Long>) PlatformDependent::usedDirectMemory);
+        Metrics.register(factory.createMetricName(DIRECT_MEMORY_LIMIT),
+                         (Gauge<Long>) PlatformDependent::maxDirectMemory);
+    }
+}

src/java/org/apache/cassandra/service/CassandraDaemon.java

@@ -82,6 +82,7 @@
 import org.apache.cassandra.locator.Locator;
 import org.apache.cassandra.metrics.CassandraMetricsRegistry;
 import org.apache.cassandra.metrics.DefaultNameFactory;
+import org.apache.cassandra.metrics.NettyMemoryMetrics;
 import org.apache.cassandra.net.StartupClusterConnectivityChecker;
 import org.apache.cassandra.schema.Schema;
 import org.apache.cassandra.schema.SchemaConstants;
@@ -352,6 +353,10 @@ protected void setup()
             }
         }
 
+        // Netty accounts for its direct memory separately from java.nio.Bits, with its own limit, so it needs its own
+        // metrics. See NettyMemoryMetrics for how this relates to gcstats and BufferPoolMetrics.
+        NettyMemoryMetrics.register();
+
         // Replay any CommitLogSegments found on disk
         PaxosState.initializeTrackers();
 

src/java/org/apache/cassandra/tools/NodeProbe.java

@@ -120,6 +120,7 @@
 import org.apache.cassandra.metrics.CQLMetrics;
 import org.apache.cassandra.metrics.CassandraMetricsRegistry;
 import org.apache.cassandra.metrics.DefaultNameFactory;
+import org.apache.cassandra.metrics.NettyMemoryMetrics;
 import org.apache.cassandra.metrics.StorageMetrics;
 import org.apache.cassandra.metrics.TableMetrics;
 import org.apache.cassandra.metrics.ThreadPoolMetrics;
@@ -1971,6 +1972,34 @@ public Object getCQLMetric(String metricName)
         }
     }
 
+    /**
+     * Retrieve Netty's own direct memory accounting, which is tracked separately from {@code java.nio.Bits} and from
+     * Cassandra's buffer pools.
+     *
+     * @param metricName UsedDirectMemory or DirectMemoryLimit
+     * @return the metric value in bytes, or -1 if Netty cannot report it
+     */
+    public Object getNettyMemoryMetric(String metricName)
+    {
+        try
+        {
+            switch (metricName)
+            {
+                case NettyMemoryMetrics.USED_DIRECT_MEMORY:
+                case NettyMemoryMetrics.DIRECT_MEMORY_LIMIT:
+                    return JMX.newMBeanProxy(mbeanServerConn,
+                                             new ObjectName("org.apache.cassandra.metrics:type=" + NettyMemoryMetrics.TYPE_NAME + ",name=" + metricName),
+                                             CassandraMetricsRegistry.JmxGaugeMBean.class).getValue();
+                default:
+                    throw new RuntimeException("Unknown NettyMemory metric name " + metricName);
+            }
+        }
+        catch (MalformedObjectNameException e)
+        {
+            throw new RuntimeException(e);
+        }
+    }
+
     private static Multimap<String, String> getJmxThreadPools(MBeanServerConnection mbeanServerConn)
     {
         try

src/java/org/apache/cassandra/tools/nodetool/Info.java

@@ -28,6 +28,7 @@
 
 import org.apache.cassandra.db.ColumnFamilyStoreMBean;
 import org.apache.cassandra.io.util.FileUtils;
+import org.apache.cassandra.metrics.NettyMemoryMetrics;
 import org.apache.cassandra.service.CacheServiceMBean;
 import org.apache.cassandra.tools.NodeProbe;
 
@@ -77,6 +78,29 @@ public void execute(NodeProbe probe)
                 throw e;
         }
 
+        // Netty tracks its direct memory separately from java.nio.Bits (reported by nodetool gcstats) and from the
+        // buffer pools reported further down, and enforces its own limit, so it gets its own line.
+        try
+        {
+            long used = (long) probe.getNettyMemoryMetric(NettyMemoryMetrics.USED_DIRECT_MEMORY);
+            long limit = (long) probe.getNettyMemoryMetric(NettyMemoryMetrics.DIRECT_MEMORY_LIMIT);
+
+            if (used < 0 || limit <= 0)
+                out.printf("%-23s: disabled%n", "Netty Direct Memory");
+            else
+                out.printf("%-23s: used %s, limit %s (%.2f%%)%n", "Netty Direct Memory",
+                           FileUtils.stringifyFileSize(used),
+                           FileUtils.stringifyFileSize(limit),
+                           used * 100.0 / limit);
+        }
+        catch (RuntimeException e)
+        {
+            if (!(e.getCause() instanceof InstanceNotFoundException))
+                throw e;
+
+            // Netty memory metrics are not registered.
+        }
+
         // Data Center/Rack
         out.printf("%-23s: %s%n", "Data Center", probe.getDataCenter());
         out.printf("%-23s: %s%n", "Rack", probe.getRack());

test/unit/org/apache/cassandra/metrics/NettyMemoryMetricsTest.java

@@ -0,0 +1,102 @@
+/*
+ * 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.cassandra.metrics;
+
+import com.codahale.metrics.Gauge;
+
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.exceptions.ConfigurationException;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.PooledByteBufAllocator;
+import io.netty.util.internal.PlatformDependent;
+
+import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics;
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class NettyMemoryMetricsTest
+{
+    @BeforeClass
+    public static void setup() throws ConfigurationException
+    {
+        DatabaseDescriptor.daemonInitialization();
+        NettyMemoryMetrics.register();
+    }
+
+    @Test
+    public void testGaugesAreRegisteredAndReflectPlatformDependent()
+    {
+        assertThat(gauge(NettyMemoryMetrics.USED_DIRECT_MEMORY).getValue())
+            .isEqualTo(PlatformDependent.usedDirectMemory());
+        assertThat(gauge(NettyMemoryMetrics.DIRECT_MEMORY_LIMIT).getValue())
+            .isEqualTo(PlatformDependent.maxDirectMemory());
+    }
+
+    @Test
+    public void testRegisterIsIdempotent()
+    {
+        Gauge<Long> before = gauge(NettyMemoryMetrics.USED_DIRECT_MEMORY);
+
+        NettyMemoryMetrics.register();
+
+        // Re-registering must not replace or duplicate the existing gauge.
+        assertThat(gauge(NettyMemoryMetrics.USED_DIRECT_MEMORY)).isSameAs(before);
+    }
+
+    @Test
+    public void testUsedDirectMemoryTracksNettyAllocations()
+    {
+        long before = gauge(NettyMemoryMetrics.USED_DIRECT_MEMORY).getValue();
+
+        // Skip when Netty cannot account for direct memory (no-cleaner path unavailable), where -1 is reported.
+        if (before < 0)
+            return;
+
+        // Allocate well beyond a single chunk so the arena has to grow rather than serve from an existing one.
+        PooledByteBufAllocator allocator = new PooledByteBufAllocator(true);
+        ByteBuf buf = allocator.directBuffer(64 * 1024 * 1024);
+        try
+        {
+            assertThat(gauge(NettyMemoryMetrics.USED_DIRECT_MEMORY).getValue()).isGreaterThan(before);
+        }
+        finally
+        {
+            buf.release();
+        }
+    }
+
+    @Test
+    public void testLimitIsPositiveWhenAccountingIsAvailable()
+    {
+        // Netty falls back to Runtime.maxMemory() when -XX:MaxDirectMemorySize is absent, so the limit should always
+        // be a usable positive number in a normal JVM.
+        assertThat(gauge(NettyMemoryMetrics.DIRECT_MEMORY_LIMIT).getValue()).isGreaterThan(0L);
+    }
+
+    @SuppressWarnings("unchecked")
+    private static Gauge<Long> gauge(String name)
+    {
+        String metricName = new DefaultNameFactory(NettyMemoryMetrics.TYPE_NAME).createMetricName(name).getMetricName();
+        Gauge<Long> gauge = (Gauge<Long>) Metrics.getGauges().get(metricName);
+        assertThat(gauge).as(metricName).isNotNull();
+        return gauge;
+    }
+}

test/unit/org/apache/cassandra/tools/nodetool/InfoTest.java

@@ -26,6 +26,7 @@
 import org.junit.Test;
 
 import org.apache.cassandra.cql3.CQLTester;
+import org.apache.cassandra.metrics.NettyMemoryMetrics;
 import org.apache.cassandra.tools.ToolRunner;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -34,12 +35,15 @@ public class InfoTest extends CQLTester
 {
     private static final Pattern PREPARED_STATEMENT_CACHE_PATTERN =
     Pattern.compile("Prepared Stmt Cache\\s+: entries (\\d+), size ([^,]+), capacity ([^,]+), (\\d+) executions, (\\d+) evictions");
+    private static final String LABEL = "Netty Direct Memory";
 
     @BeforeClass
     public static void setup() throws Exception
     {
         requireNetwork();
         startJMXServer();
+        // Normally registered by CassandraDaemon.setup(), which does not run for in-JVM tests.
+        NettyMemoryMetrics.register();
     }
 
     @Test
@@ -59,4 +63,25 @@ public void testInfoContainsPreparedStatementCache()
         assertThat(matcher.group(2)).isNotEqualTo("0 bytes");
         assertThat(Integer.parseInt(matcher.group(4))).isGreaterThan(0);
     }
+
+    @Test
+    public void testInfoReportsNettyDirectMemory()
+    {
+        ToolRunner.ToolResult tool = ToolRunner.invokeNodetool("info");
+        tool.assertOnCleanExit();
+
+        String output = tool.getStdout();
+        assertThat(output).contains(LABEL);
+
+        // Either Netty's direct memory accounting is disabled, or we get used/limit plus a well formed percentage.
+        Matcher matcher = Pattern.compile(LABEL + "\\s*: used .+, limit .+ \\(([0-9.]+)%\\)").matcher(output);
+        if (!matcher.find())
+        {
+            assertThat(output).containsPattern(LABEL + "\\s*: disabled");
+            return;
+        }
+
+        double percentage = Double.parseDouble(matcher.group(1));
+        assertThat(percentage).isBetween(0.0d, 100.0d);
+    }
 }