[rpc] Support netty metrics (#407)

SteNicholas · web-flow · commit 99fd922d632e · 2025-02-25T16:26:34.000+08:00
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/metrics/MetricNames.java b/fluss-common/src/main/java/com/alibaba/fluss/metrics/MetricNames.java
@@ -152,4 +152,11 @@ public class MetricNames {
     public static final String SCANNER_REMOTE_FETCH_BYTES_RATE = "remoteFetchBytesPerSecond";
     public static final String SCANNER_REMOTE_FETCH_RATE = "remoteFetchRequestsPerSecond";
     public static final String SCANNER_REMOTE_FETCH_ERROR_RATE = "remoteFetchErrorPerSecond";
+
+    // for netty
+    public static final String NETTY_USED_DIRECT_MEMORY = "usedDirectMemory";
+    public static final String NETTY_NUM_DIRECT_ARENAS = "numDirectArenas";
+    public static final String NETTY_NUM_ALLOCATIONS_PER_SECONDS = "numAllocationsPerSecond";
+    public static final String NETTY_NUM_HUGE_ALLOCATIONS_PER_SECONDS =
+            "numHugeAllocationsPerSecond";
 }
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/metrics/groups/AbstractMetricGroup.java b/fluss-common/src/main/java/com/alibaba/fluss/metrics/groups/AbstractMetricGroup.java
@@ -17,6 +17,7 @@
 package com.alibaba.fluss.metrics.groups;
 
 import com.alibaba.fluss.annotation.Internal;
+import com.alibaba.fluss.annotation.VisibleForTesting;
 import com.alibaba.fluss.metrics.CharacterFilter;
 import com.alibaba.fluss.metrics.Counter;
 import com.alibaba.fluss.metrics.Gauge;
@@ -361,6 +362,11 @@ protected GenericMetricGroup createChildGroup(String name, ChildType childType)
         }
     }
 
+    @VisibleForTesting
+    public Map<String, Metric> getMetrics() {
+        return metrics;
+    }
+
     /**
      * Enum for indicating which child group should be created. `KEY` is used to create {@link
      * GenericKeyMetricGroup}. `VALUE` is used to create {@link GenericValueMetricGroup}. `GENERIC`
diff --git a/fluss-rpc/src/main/java/com/alibaba/fluss/rpc/RpcServer.java b/fluss-rpc/src/main/java/com/alibaba/fluss/rpc/RpcServer.java
@@ -17,6 +17,7 @@
 package com.alibaba.fluss.rpc;
 
 import com.alibaba.fluss.config.Configuration;
+import com.alibaba.fluss.metrics.groups.MetricGroup;
 import com.alibaba.fluss.rpc.netty.server.NettyServer;
 import com.alibaba.fluss.rpc.netty.server.RequestsMetrics;
 import com.alibaba.fluss.utils.AutoCloseableAsync;
@@ -36,6 +37,7 @@ public interface RpcServer extends AutoCloseableAsync {
      * @param externalAddress The external address to bind to.
      * @param externalPortRange The external port range to bind to.
      * @param service The service to handle incoming requests.
+     * @param serverMetricGroup The metric group of server to report.
      * @param requestsMetrics the requests metrics to report.
      * @return The new RPC server.
      */
@@ -44,9 +46,16 @@ static RpcServer create(
             String externalAddress,
             String externalPortRange,
             RpcGatewayService service,
+            MetricGroup serverMetricGroup,
             RequestsMetrics requestsMetrics)
             throws IOException {
-        return new NettyServer(conf, externalAddress, externalPortRange, service, requestsMetrics);
+        return new NettyServer(
+                conf,
+                externalAddress,
+                externalPortRange,
+                service,
+                serverMetricGroup,
+                requestsMetrics);
     }
 
     /** Starts the RPC server by binding to the configured bind address and port (blocking). */
diff --git a/fluss-rpc/src/main/java/com/alibaba/fluss/rpc/netty/NettyMetrics.java b/fluss-rpc/src/main/java/com/alibaba/fluss/rpc/netty/NettyMetrics.java
@@ -0,0 +1,55 @@
+/*
+ * Copyright (c) 2024 Alibaba Group Holding Ltd.
+ *
+ * Licensed 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 com.alibaba.fluss.rpc.netty;
+
+import com.alibaba.fluss.metrics.Gauge;
+import com.alibaba.fluss.metrics.MeterView;
+import com.alibaba.fluss.metrics.MetricNames;
+import com.alibaba.fluss.metrics.groups.MetricGroup;
+import com.alibaba.fluss.shaded.netty4.io.netty.buffer.PoolArenaMetric;
+import com.alibaba.fluss.shaded.netty4.io.netty.buffer.PooledByteBufAllocator;
+import com.alibaba.fluss.shaded.netty4.io.netty.buffer.PooledByteBufAllocatorMetric;
+
+/** A netty metrics class to register metrics from netty. */
+public class NettyMetrics {
+
+    public static final String NETTY_METRIC_GROUP = "netty";
+
+    public static void registerNettyMetrics(
+            MetricGroup metricGroup, PooledByteBufAllocator pooledAllocator) {
+        MetricGroup nettyMetricGroup = metricGroup.addGroup(NETTY_METRIC_GROUP);
+        PooledByteBufAllocatorMetric pooledAllocatorMetric = pooledAllocator.metric();
+        nettyMetricGroup.<Long, Gauge<Long>>gauge(
+                MetricNames.NETTY_USED_DIRECT_MEMORY, pooledAllocatorMetric::usedDirectMemory);
+        nettyMetricGroup.<Integer, Gauge<Integer>>gauge(
+                MetricNames.NETTY_NUM_DIRECT_ARENAS, pooledAllocatorMetric::numDirectArenas);
+        nettyMetricGroup.meter(
+                MetricNames.NETTY_NUM_ALLOCATIONS_PER_SECONDS,
+                new MeterView(
+                        () ->
+                                pooledAllocatorMetric.directArenas().stream()
+                                        .mapToLong(PoolArenaMetric::numAllocations)
+                                        .sum()));
+        nettyMetricGroup.meter(
+                MetricNames.NETTY_NUM_HUGE_ALLOCATIONS_PER_SECONDS,
+                new MeterView(
+                        () ->
+                                pooledAllocatorMetric.directArenas().stream()
+                                        .mapToLong(PoolArenaMetric::numHugeAllocations)
+                                        .sum()));
+    }
+}
diff --git a/fluss-rpc/src/main/java/com/alibaba/fluss/rpc/netty/client/NettyClient.java b/fluss-rpc/src/main/java/com/alibaba/fluss/rpc/netty/client/NettyClient.java
@@ -23,6 +23,7 @@
 import com.alibaba.fluss.rpc.RpcClient;
 import com.alibaba.fluss.rpc.messages.ApiMessage;
 import com.alibaba.fluss.rpc.metrics.ClientMetricGroup;
+import com.alibaba.fluss.rpc.netty.NettyMetrics;
 import com.alibaba.fluss.rpc.netty.NettyUtils;
 import com.alibaba.fluss.rpc.protocol.ApiKeys;
 import com.alibaba.fluss.shaded.netty4.io.netty.bootstrap.Bootstrap;
@@ -82,14 +83,16 @@ public NettyClient(Configuration conf, ClientMetricGroup clientMetricGroup) {
         int connectTimeoutMs = (int) conf.get(ConfigOptions.CLIENT_CONNECT_TIMEOUT).toMillis();
         int connectionMaxIdle =
                 (int) conf.get(ConfigOptions.NETTY_CONNECTION_MAX_IDLE_TIME).getSeconds();
+        PooledByteBufAllocator pooledAllocator = PooledByteBufAllocator.DEFAULT;
         this.bootstrap =
                 new Bootstrap()
                         .group(eventGroup)
                         .channel(NettyUtils.getClientSocketChannelClass(eventGroup))
-                        .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
+                        .option(ChannelOption.ALLOCATOR, pooledAllocator)
                         .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutMs)
                         .handler(new ClientChannelInitializer(connectionMaxIdle));
         this.clientMetricGroup = clientMetricGroup;
+        NettyMetrics.registerNettyMetrics(clientMetricGroup, pooledAllocator);
     }
 
     /**
diff --git a/fluss-rpc/src/main/java/com/alibaba/fluss/rpc/netty/server/NettyServer.java b/fluss-rpc/src/main/java/com/alibaba/fluss/rpc/netty/server/NettyServer.java
@@ -18,9 +18,11 @@
 
 import com.alibaba.fluss.config.ConfigOptions;
 import com.alibaba.fluss.config.Configuration;
+import com.alibaba.fluss.metrics.groups.MetricGroup;
 import com.alibaba.fluss.rpc.RpcGateway;
 import com.alibaba.fluss.rpc.RpcGatewayService;
 import com.alibaba.fluss.rpc.RpcServer;
+import com.alibaba.fluss.rpc.netty.NettyMetrics;
 import com.alibaba.fluss.rpc.netty.NettyUtils;
 import com.alibaba.fluss.rpc.protocol.ApiManager;
 import com.alibaba.fluss.shaded.netty4.io.netty.bootstrap.ServerBootstrap;
@@ -60,6 +62,7 @@ public final class NettyServer implements RpcServer {
     private final String portRange;
     private final RequestProcessorPool workerPool;
     private final ApiManager apiManager;
+    private final MetricGroup serverMetricGroup;
 
     private EventLoopGroup acceptorGroup;
     private EventLoopGroup selectorGroup;
@@ -73,10 +76,12 @@ public NettyServer(
             String hostname,
             String portRange,
             RpcGatewayService service,
+            MetricGroup serverMetricGroup,
             RequestsMetrics requestsMetrics) {
         this.conf = checkNotNull(conf, "conf");
         this.hostname = checkNotNull(hostname, "hostname");
         this.portRange = checkNotNull(portRange, "portRange");
+        this.serverMetricGroup = checkNotNull(serverMetricGroup, "serverMetricGroup");
         this.apiManager = new ApiManager(service.providerType());
 
         this.workerPool =
@@ -109,7 +114,8 @@ public void start() throws IOException {
                 NettyUtils.newEventLoopGroup(numNetworkThreads, "fluss-netty-server-selector");
 
         ServerBootstrap bootstrap = new ServerBootstrap();
-        bootstrap.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
+        PooledByteBufAllocator pooledBufAllocator = PooledByteBufAllocator.DEFAULT;
+        bootstrap.childOption(ChannelOption.ALLOCATOR, pooledBufAllocator);
         bootstrap.group(acceptorGroup, selectorGroup);
         bootstrap.childOption(ChannelOption.TCP_NODELAY, true);
         bootstrap.childOption(
@@ -168,6 +174,7 @@ public void start() throws IOException {
                 bindAddress);
 
         isRunning = true;
+        NettyMetrics.registerNettyMetrics(serverMetricGroup, pooledBufAllocator);
     }
 
     @Override
diff --git a/fluss-rpc/src/test/java/com/alibaba/fluss/rpc/netty/NettyMetricsTest.java b/fluss-rpc/src/test/java/com/alibaba/fluss/rpc/netty/NettyMetricsTest.java
@@ -0,0 +1,99 @@
+/*
+ * Copyright (c) 2024 Alibaba Group Holding Ltd.
+ *
+ * Licensed 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 com.alibaba.fluss.rpc.netty;
+
+import com.alibaba.fluss.cluster.ServerNode;
+import com.alibaba.fluss.cluster.ServerType;
+import com.alibaba.fluss.config.Configuration;
+import com.alibaba.fluss.metrics.Metric;
+import com.alibaba.fluss.metrics.MetricNames;
+import com.alibaba.fluss.metrics.groups.GenericMetricGroup;
+import com.alibaba.fluss.metrics.groups.MetricGroup;
+import com.alibaba.fluss.metrics.util.NOPMetricsGroup;
+import com.alibaba.fluss.rpc.TestingGatewayService;
+import com.alibaba.fluss.rpc.metrics.ClientMetricGroup;
+import com.alibaba.fluss.rpc.metrics.TestingClientMetricGroup;
+import com.alibaba.fluss.rpc.netty.client.NettyClient;
+import com.alibaba.fluss.rpc.netty.server.NettyServer;
+import com.alibaba.fluss.rpc.netty.server.RequestsMetrics;
+import com.alibaba.fluss.utils.NetUtils;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+import static com.alibaba.fluss.utils.NetUtils.getAvailablePort;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link NettyMetrics}. */
+public class NettyMetricsTest {
+
+    private NettyServer nettyServer;
+    private NettyClient nettyClient;
+    private MetricGroup serverGroup;
+    private MetricGroup clientGroup;
+
+    @BeforeEach
+    public void setup() throws Exception {
+        Configuration conf = new Configuration();
+        try (NetUtils.Port availablePort = getAvailablePort()) {
+            ServerNode serverNode =
+                    new ServerNode(1, "localhost", availablePort.getPort(), ServerType.COORDINATOR);
+            MetricGroup serverMetricGroup = NOPMetricsGroup.newInstance();
+            serverGroup = serverMetricGroup.addGroup(NettyMetrics.NETTY_METRIC_GROUP);
+            nettyServer =
+                    new NettyServer(
+                            conf,
+                            serverNode.host(),
+                            String.valueOf(serverNode.port()),
+                            new TestingGatewayService(),
+                            serverMetricGroup,
+                            RequestsMetrics.createCoordinatorServerRequestMetrics(
+                                    serverMetricGroup));
+            nettyServer.start();
+        }
+        ClientMetricGroup clientMetricGroup = TestingClientMetricGroup.newInstance();
+        clientGroup = clientMetricGroup.addGroup(NettyMetrics.NETTY_METRIC_GROUP);
+        nettyClient = new NettyClient(conf, clientMetricGroup);
+    }
+
+    @AfterEach
+    public void cleanup() throws Exception {
+        if (nettyServer != null) {
+            nettyServer.close();
+        }
+        if (nettyClient != null) {
+            nettyClient.close();
+        }
+    }
+
+    @Test
+    void testNettyMetricsCompleteness() {
+        assertNettyMetrics(serverGroup);
+        assertNettyMetrics(clientGroup);
+    }
+
+    private void assertNettyMetrics(MetricGroup metricGroup) {
+        Map<String, Metric> metrics = ((GenericMetricGroup) metricGroup).getMetrics();
+        assertThat(metrics.get(MetricNames.NETTY_USED_DIRECT_MEMORY)).isNotNull();
+        assertThat(metrics.get(MetricNames.NETTY_NUM_DIRECT_ARENAS)).isNotNull();
+        assertThat(metrics.get(MetricNames.NETTY_NUM_ALLOCATIONS_PER_SECONDS)).isNotNull();
+        assertThat(metrics.get(MetricNames.NETTY_NUM_HUGE_ALLOCATIONS_PER_SECONDS)).isNotNull();
+    }
+}
diff --git a/fluss-rpc/src/test/java/com/alibaba/fluss/rpc/netty/client/NettyClientTest.java b/fluss-rpc/src/test/java/com/alibaba/fluss/rpc/netty/client/NettyClientTest.java
@@ -20,6 +20,7 @@
 import com.alibaba.fluss.cluster.ServerType;
 import com.alibaba.fluss.config.ConfigOptions;
 import com.alibaba.fluss.config.Configuration;
+import com.alibaba.fluss.metrics.groups.MetricGroup;
 import com.alibaba.fluss.metrics.util.NOPMetricsGroup;
 import com.alibaba.fluss.rpc.TestingGatewayService;
 import com.alibaba.fluss.rpc.messages.ApiMessage;
@@ -186,14 +187,15 @@ private void buildNettyServer(int serverId) throws Exception {
                     new ServerNode(
                             serverId, "localhost", availablePort.getPort(), ServerType.COORDINATOR);
             service = new TestingGatewayService();
+            MetricGroup metricGroup = NOPMetricsGroup.newInstance();
             nettyServer =
                     new NettyServer(
                             conf,
                             serverNode.host(),
                             String.valueOf(serverNode.port()),
                             service,
-                            RequestsMetrics.createCoordinatorServerRequestMetrics(
-                                    NOPMetricsGroup.newInstance()));
+                            metricGroup,
+                            RequestsMetrics.createCoordinatorServerRequestMetrics(metricGroup));
             nettyServer.start();
         }
     }
diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/coordinator/CoordinatorServer.java b/fluss-server/src/main/java/com/alibaba/fluss/server/coordinator/CoordinatorServer.java
@@ -156,6 +156,7 @@ protected void startServices() throws Exception {
                             conf.getString(ConfigOptions.COORDINATOR_HOST),
                             conf.getString(ConfigOptions.COORDINATOR_PORT),
                             coordinatorService,
+                            serverMetricGroup,
                             RequestsMetrics.createCoordinatorServerRequestMetrics(
                                     serverMetricGroup));
             rpcServer.start();
diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/tablet/TabletServer.java b/fluss-server/src/main/java/com/alibaba/fluss/server/tablet/TabletServer.java
@@ -209,6 +209,7 @@ protected void startServices() throws Exception {
                             conf.getString(ConfigOptions.TABLET_SERVER_HOST),
                             conf.getString(ConfigOptions.TABLET_SERVER_PORT),
                             tabletService,
+                            tabletServerMetricGroup,
                             requestsMetrics);
             rpcServer.start();
 
diff --git a/website/docs/maintenance/observability/monitor-metrics.md b/website/docs/maintenance/observability/monitor-metrics.md
@@ -242,6 +242,44 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM
   </tbody>
 </table>
 
+### Netty
+
+<table class="table table-bordered">
+  <thead>
+    <tr>
+      <th class="text-left" style={{width: '30pt'}}>Scope</th>
+      <th class="text-left" style={{width: '150pt'}}>Infix</th>
+      <th class="text-left" style={{width: '80pt'}}>Metrics</th>
+      <th class="text-left" style={{width: '300pt'}}>Description</th>
+      <th class="text-left" style={{width: '40pt'}}>Type</th>
+    </tr>
+  </thead>
+  <tbody>
+    <tr>
+      <th rowspan="1"><strong>coordinator/tabletserver/client</strong></th>
+      <td rowspan="1">netty</td>
+      <td>usedDirectMemory</td>
+      <td>The number of bytes of direct memory used by netty.</td>
+      <td>Gauge</td>
+    </tr>
+    <tr>
+      <td>numDirectArenas</td>
+      <td>The number of direct arenas in netty.</td>
+      <td>Gauge</td>
+    </tr>
+    <tr>
+      <td>numAllocationsPerSecond</td>
+      <td>The number of allocations done via the arena per second.</td>
+      <td>Meter</td>
+    </tr>
+    <tr>
+      <td>numHugeAllocationsPerSecond</td>
+      <td>The number of huge allocations done via the arena per second.</td>
+      <td>Meter</td>
+    </tr>
+  </tbody>
+</table>
+
 ### Coordinator Server
 
 <table class="table table-bordered">
