Merge branch 'dev' into improve_e2e_m1_chip_local_mode

docs/docs/en/guide/metrics/metrics.md

@@ -91,6 +91,11 @@ For example, you can get the master metrics by `curl http://localhost:5679/actua
   - stop: the number of stopped workflow instances
   - failover: the number of workflow instance fail-overs
 
+### RPC Related Metrics
+
+- ds.rpc.client.sync.request.exception.count: (counter) the number of exceptions occurred in sync rpc requests
+- ds.rpc.client.sync.request.duration.time: (histogram) the time cost of sync rpc requests
+
 ### Master Server Metrics
 
 - ds.master.overload.count: (counter) the number of times the master overloaded

docs/docs/zh/guide/metrics/metrics.md

@@ -91,6 +91,11 @@ metrics exporter端口`server.port`是在application.yaml里定义的: master: `
   - stop：停止的工作流实例数量
   - failover：容错的工作流实例数量
 
+### RPC相关指标
+
+- ds.rpc.client.sync.request.exception.count: (counter) 同步rpc请求异常数
+- ds.rpc.client.sync.request.duration.time: (histogram) 同步rpc请求耗时
+
 ### Master Server指标
 
 - ds.master.overload.count: (counter) master过载次数

dolphinscheduler-extract/dolphinscheduler-extract-base/pom.xml

@@ -47,6 +47,13 @@
             <artifactId>dolphinscheduler-common</artifactId>
             <version>${project.version}</version>
         </dependency>
+
+        <dependency>
+            <groupId>org.apache.dolphinscheduler</groupId>
+            <artifactId>dolphinscheduler-meter</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
         <dependency>
             <groupId>io.netty</groupId>
             <artifactId>netty-all</artifactId>

dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/RpcMethod.java

@@ -28,6 +28,6 @@ import java.lang.annotation.Target;
 @Documented
 public @interface RpcMethod {
 
-    long timeout() default 3000L;
+    long timeout() default -1;
 
 }

dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/SyncRequestDto.java

@@ -0,0 +1,38 @@
+/*
+ * 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.dolphinscheduler.extract.base;
+
+import org.apache.dolphinscheduler.extract.base.protocal.Transporter;
+import org.apache.dolphinscheduler.extract.base.utils.Host;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
+public class SyncRequestDto {
+
+    private Host serverHost;
+    private Transporter transporter;
+    private long timeoutMillis;
+
+}

dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/client/NettyClientHandler.java

@@ -77,7 +77,7 @@ public class NettyClientHandler extends ChannelInboundHandlerAdapter {
                     .writeAndFlush(HeartBeatTransporter.getHeartBeatTransporter())
                     .addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
             if (log.isDebugEnabled()) {
-                log.debug("Client send heart beat to: {}", ChannelUtils.getRemoteAddress(ctx.channel()));
+                log.info("Client send heartbeat to: {}", ctx.channel().remoteAddress());
             }
         } else {
             super.userEventTriggered(ctx, evt);

dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/client/NettyRemotingClient.java

@@ -19,14 +19,17 @@ package org.apache.dolphinscheduler.extract.base.client;
 
 import org.apache.dolphinscheduler.common.thread.ThreadUtils;
 import org.apache.dolphinscheduler.extract.base.IRpcResponse;
+import org.apache.dolphinscheduler.extract.base.SyncRequestDto;
 import org.apache.dolphinscheduler.extract.base.config.NettyClientConfig;
 import org.apache.dolphinscheduler.extract.base.exception.RemotingException;
 import org.apache.dolphinscheduler.extract.base.exception.RemotingTimeoutException;
 import org.apache.dolphinscheduler.extract.base.future.ResponseFuture;
+import org.apache.dolphinscheduler.extract.base.metrics.ClientSyncDurationMetrics;
+import org.apache.dolphinscheduler.extract.base.metrics.ClientSyncExceptionMetrics;
+import org.apache.dolphinscheduler.extract.base.metrics.RpcMetrics;
 import org.apache.dolphinscheduler.extract.base.protocal.Transporter;
 import org.apache.dolphinscheduler.extract.base.protocal.TransporterDecoder;
 import org.apache.dolphinscheduler.extract.base.protocal.TransporterEncoder;
-import org.apache.dolphinscheduler.extract.base.utils.Constants;
 import org.apache.dolphinscheduler.extract.base.utils.Host;
 import org.apache.dolphinscheduler.extract.base.utils.NettyUtils;
 
@@ -97,8 +100,8 @@ public class NettyRemotingClient implements AutoCloseable {
                         ch.pipeline()
                                 .addLast("client-idle-handler",
                                         new IdleStateHandler(
-                                                Constants.NETTY_CLIENT_HEART_BEAT_TIME,
                                                 0,
+                                                clientConfig.getHeartBeatIntervalMillis(),
                                                 0,
                                                 TimeUnit.MILLISECONDS))
                                 .addLast(new TransporterDecoder(), clientHandler, new TransporterEncoder());
@@ -107,38 +110,60 @@ public class NettyRemotingClient implements AutoCloseable {
         isStarted.compareAndSet(false, true);
     }
 
-    public IRpcResponse sendSync(final Host host,
-                                 final Transporter transporter,
-                                 final long timeoutMillis) throws InterruptedException, RemotingException {
-        final Channel channel = getOrCreateChannel(host);
-        if (channel == null) {
-            throw new RemotingException(String.format("connect to : %s fail", host));
-        }
+    public IRpcResponse sendSync(SyncRequestDto syncRequestDto) throws RemotingException {
+        long start = System.currentTimeMillis();
+
+        final Host host = syncRequestDto.getServerHost();
+        final Transporter transporter = syncRequestDto.getTransporter();
+        final long timeoutMillis = syncRequestDto.getTimeoutMillis() < 0 ? clientConfig.getConnectTimeoutMillis()
+                : syncRequestDto.getTimeoutMillis();
         final long opaque = transporter.getHeader().getOpaque();
-        final ResponseFuture responseFuture = new ResponseFuture(opaque, timeoutMillis);
-        channel.writeAndFlush(transporter).addListener(future -> {
-            if (future.isSuccess()) {
-                responseFuture.setSendOk(true);
-                return;
-            } else {
-                responseFuture.setSendOk(false);
+
+        try {
+            final Channel channel = getOrCreateChannel(host);
+            if (channel == null) {
+                throw new RemotingException(String.format("connect to : %s fail", host));
             }
-            responseFuture.setCause(future.cause());
-            responseFuture.putResponse(null);
-            log.error("Send Sync request {} to host {} failed", transporter, host, responseFuture.getCause());
-        });
-        /*
-         * sync wait for result
-         */
-        IRpcResponse iRpcResponse = responseFuture.waitResponse();
-        if (iRpcResponse == null) {
-            if (responseFuture.isSendOK()) {
-                throw new RemotingTimeoutException(host.toString(), timeoutMillis, responseFuture.getCause());
-            } else {
-                throw new RemotingException(host.toString(), responseFuture.getCause());
+            final ResponseFuture responseFuture = new ResponseFuture(opaque, timeoutMillis);
+            channel.writeAndFlush(transporter).addListener(future -> {
+                if (future.isSuccess()) {
+                    responseFuture.setSendOk(true);
+                    return;
+                } else {
+                    responseFuture.setSendOk(false);
+                }
+                responseFuture.setCause(future.cause());
+                responseFuture.putResponse(null);
+                log.error("Send Sync request {} to host {} failed", transporter, host, responseFuture.getCause());
+            });
+            /*
+             * sync wait for result
+             */
+            IRpcResponse iRpcResponse = responseFuture.waitResponse();
+            if (iRpcResponse == null) {
+                if (responseFuture.isSendOK()) {
+                    throw new RemotingTimeoutException(host.toString(), timeoutMillis, responseFuture.getCause());
+                } else {
+                    throw new RemotingException(host.toString(), responseFuture.getCause());
+                }
             }
+            return iRpcResponse;
+        } catch (Exception ex) {
+            ClientSyncExceptionMetrics clientSyncExceptionMetrics = ClientSyncExceptionMetrics
+                    .of(syncRequestDto)
+                    .withThrowable(ex);
+            RpcMetrics.recordClientSyncRequestException(clientSyncExceptionMetrics);
+            if (ex instanceof RemotingException) {
+                throw (RemotingException) ex;
+            } else {
+                throw new RemotingException(ex);
+            }
+        } finally {
+            ClientSyncDurationMetrics clientSyncDurationMetrics = ClientSyncDurationMetrics
+                    .of(syncRequestDto)
+                    .withMilliseconds(System.currentTimeMillis() - start);
+            RpcMetrics.recordClientSyncRequestDuration(clientSyncDurationMetrics);
         }
-        return iRpcResponse;
     }
 
     Channel getOrCreateChannel(Host host) {

dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/client/SyncClientMethodInvoker.java

@@ -20,6 +20,7 @@ package org.apache.dolphinscheduler.extract.base.client;
 import org.apache.dolphinscheduler.extract.base.IRpcResponse;
 import org.apache.dolphinscheduler.extract.base.RpcMethod;
 import org.apache.dolphinscheduler.extract.base.StandardRpcRequest;
+import org.apache.dolphinscheduler.extract.base.SyncRequestDto;
 import org.apache.dolphinscheduler.extract.base.exception.MethodInvocationException;
 import org.apache.dolphinscheduler.extract.base.protocal.Transporter;
 import org.apache.dolphinscheduler.extract.base.protocal.TransporterHeader;
@@ -41,8 +42,12 @@ class SyncClientMethodInvoker extends AbstractClientMethodInvoker {
         transporter.setBody(JsonSerializer.serialize(StandardRpcRequest.of(args)));
         transporter.setHeader(TransporterHeader.of(methodIdentifier));
 
-        IRpcResponse iRpcResponse =
-                nettyRemotingClient.sendSync(serverHost, transporter, sync.timeout());
+        SyncRequestDto syncRequestDto = SyncRequestDto.builder()
+                .timeoutMillis(sync.timeout())
+                .transporter(transporter)
+                .serverHost(serverHost)
+                .build();
+        IRpcResponse iRpcResponse = nettyRemotingClient.sendSync(syncRequestDto);
         if (!iRpcResponse.isSuccess()) {
             throw MethodInvocationException.of(iRpcResponse.getMessage());
         }

dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/config/NettyClientConfig.java

@@ -17,6 +17,8 @@
 
 package org.apache.dolphinscheduler.extract.base.config;
 
+import java.time.Duration;
+
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
@@ -64,4 +66,14 @@ public class NettyClientConfig {
     @Builder.Default
     private int connectTimeoutMillis = 3000;
 
+    /**
+     * Will send {@link org.apache.dolphinscheduler.extract.base.protocal.HeartBeatTransporter} to netty server every
+     * heartBeatIntervalMillis, used to keep the {@link io.netty.channel.Channel} active.
+     */
+    @Builder.Default
+    private long heartBeatIntervalMillis = Duration.ofSeconds(10).toMillis();
+
+    @Builder.Default
+    private int defaultRpcTimeoutMillis = 10_000;
+
 }

dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/config/NettyServerConfig.java

@@ -17,6 +17,8 @@
 
 package org.apache.dolphinscheduler.extract.base.config;
 
+import java.time.Duration;
+
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
@@ -66,6 +68,12 @@ public class NettyServerConfig {
     @Builder.Default
     private int workerThread = Runtime.getRuntime().availableProcessors() * 2;
 
+    /**
+     * If done's receive any data from a {@link io.netty.channel.Channel} during 180s then will close it.
+     */
+    @Builder.Default
+    private long connectionIdleTime = Duration.ofSeconds(60).toMillis();
+
     /**
      * listen port
      */

dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/metrics/ClientSyncDurationMetrics.java

@@ -0,0 +1,56 @@
+/*
+ * 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.dolphinscheduler.extract.base.metrics;
+
+import org.apache.dolphinscheduler.common.utils.NetUtils;
+import org.apache.dolphinscheduler.extract.base.SyncRequestDto;
+import org.apache.dolphinscheduler.extract.base.protocal.Transporter;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class ClientSyncDurationMetrics {
+
+    private Transporter transporter;
+
+    private long milliseconds;
+
+    @Builder.Default
+    private String clientHost = NetUtils.getHost();
+
+    private String serverHost;
+
+    public static ClientSyncDurationMetrics of(SyncRequestDto syncRequestDto) {
+        return ClientSyncDurationMetrics.builder()
+                .transporter(syncRequestDto.getTransporter())
+                .serverHost(syncRequestDto.getServerHost().getIp())
+                .build();
+    }
+
+    public ClientSyncDurationMetrics withMilliseconds(long milliseconds) {
+        this.milliseconds = milliseconds;
+        return this;
+    }
+
+}

dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/metrics/ClientSyncExceptionMetrics.java

@@ -0,0 +1,56 @@
+/*
+ * 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.dolphinscheduler.extract.base.metrics;
+
+import org.apache.dolphinscheduler.common.utils.NetUtils;
+import org.apache.dolphinscheduler.extract.base.SyncRequestDto;
+import org.apache.dolphinscheduler.extract.base.protocal.Transporter;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class ClientSyncExceptionMetrics {
+
+    private Transporter transporter;
+
+    private String clientHost;
+
+    @Builder.Default
+    private String serverHost = NetUtils.getHost();
+
+    private Throwable throwable;
+
+    public static ClientSyncExceptionMetrics of(SyncRequestDto syncRequestDto) {
+        return ClientSyncExceptionMetrics.builder()
+                .transporter(syncRequestDto.getTransporter())
+                .build();
+
+    }
+
+    public ClientSyncExceptionMetrics withThrowable(Throwable throwable) {
+        this.throwable = throwable;
+        return this;
+    }
+
+}

dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/metrics/RpcMetrics.java

@@ -0,0 +1,94 @@
+/*
+ * 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.dolphinscheduler.extract.base.metrics;
+
+import org.apache.dolphinscheduler.extract.base.protocal.Transporter;
+import org.apache.dolphinscheduler.extract.base.protocal.TransporterHeader;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.Metrics;
+import io.micrometer.core.instrument.Timer;
+
+public class RpcMetrics {
+
+    private static final Map<String, Timer> rpcRequestDurationTimer = new ConcurrentHashMap<>();
+
+    private static final Map<String, Counter> rpcRequestExceptionCounter = new ConcurrentHashMap<>();
+
+    public static void recordClientSyncRequestException(ClientSyncExceptionMetrics clientSyncExceptionMetrics) {
+        recordClientSyncRequestException(
+                clientSyncExceptionMetrics.getThrowable(),
+                Optional.of(clientSyncExceptionMetrics)
+                        .map(ClientSyncExceptionMetrics::getTransporter)
+                        .map(Transporter::getHeader)
+                        .map(TransporterHeader::getMethodIdentifier)
+                        .orElseGet(() -> "unknown"),
+                clientSyncExceptionMetrics.getClientHost(),
+                clientSyncExceptionMetrics.getServerHost());
+    }
+
+    public static void recordClientSyncRequestException(final Throwable throwable,
+                                                        final String methodName,
+                                                        final String clientHost,
+                                                        final String serverHost) {
+        final String exceptionType = throwable == null ? "unknown" : throwable.getClass().getSimpleName();
+        final Counter counter = rpcRequestExceptionCounter.computeIfAbsent(exceptionType,
+                (et) -> Counter.builder("ds.rpc.client.sync.request.exception.count")
+                        .tag("method_name", methodName)
+                        .tag("client_host", clientHost)
+                        .tag("server_host", serverHost)
+                        .tag("exception_name", et)
+                        .description("rpc sync request exception counter for exception type: " + et)
+                        .register(Metrics.globalRegistry));
+        counter.increment();
+    }
+
+    public static void recordClientSyncRequestDuration(ClientSyncDurationMetrics clientSyncDurationMetrics) {
+        recordClientSyncRequestDuration(
+                Optional.of(clientSyncDurationMetrics)
+                        .map(ClientSyncDurationMetrics::getTransporter)
+                        .map(Transporter::getHeader)
+                        .map(TransporterHeader::getMethodIdentifier)
+                        .orElseGet(() -> "unknown"),
+                clientSyncDurationMetrics.getMilliseconds(),
+                clientSyncDurationMetrics.getClientHost(),
+                clientSyncDurationMetrics.getServerHost());
+    }
+
+    public static void recordClientSyncRequestDuration(final String methodName,
+                                                       final long milliseconds,
+                                                       final String clientHost,
+                                                       final String serverHost) {
+        rpcRequestDurationTimer.computeIfAbsent(methodName,
+                (method) -> Timer.builder("ds.rpc.client.sync.request.duration.time")
+                        .tag("method_name", method)
+                        .tag("client_host", clientHost)
+                        .tag("server_host", serverHost)
+                        .publishPercentiles(0.5, 0.75, 0.95, 0.99)
+                        .publishPercentileHistogram()
+                        .description("time cost of sync rpc request, unit ms")
+                        .register(Metrics.globalRegistry))
+                .record(milliseconds, TimeUnit.MILLISECONDS);
+    }
+
+}

dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/server/JdkDynamicServerHandler.java

@@ -38,6 +38,7 @@ import io.netty.channel.ChannelConfig;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.handler.timeout.IdleState;
 import io.netty.handler.timeout.IdleStateEvent;
 
 @Slf4j
@@ -160,7 +161,11 @@ class JdkDynamicServerHandler extends ChannelInboundHandlerAdapter {
     @Override
     public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
         if (evt instanceof IdleStateEvent) {
-            ctx.channel().close();
+            IdleStateEvent event = (IdleStateEvent) evt;
+            if (event.state() == IdleState.READER_IDLE) {
+                log.warn("Not receive heart beat from: {}, will close the channel", ctx.channel().remoteAddress());
+                ctx.close();
+            }
         } else {
             super.userEventTriggered(ctx, evt);
         }

dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/server/NettyRemotingServer.java

@@ -22,7 +22,6 @@ import org.apache.dolphinscheduler.extract.base.config.NettyServerConfig;
 import org.apache.dolphinscheduler.extract.base.exception.RemoteException;
 import org.apache.dolphinscheduler.extract.base.protocal.TransporterDecoder;
 import org.apache.dolphinscheduler.extract.base.protocal.TransporterEncoder;
-import org.apache.dolphinscheduler.extract.base.utils.Constants;
 import org.apache.dolphinscheduler.extract.base.utils.NettyUtils;
 
 import java.util.concurrent.ExecutorService;
@@ -135,7 +134,7 @@ class NettyRemotingServer {
                 .addLast("encoder", new TransporterEncoder())
                 .addLast("decoder", new TransporterDecoder())
                 .addLast("server-idle-handle",
-                        new IdleStateHandler(0, 0, Constants.NETTY_SERVER_HEART_BEAT_TIME, TimeUnit.MILLISECONDS))
+                        new IdleStateHandler(serverConfig.getConnectionIdleTime(), 0, 0, TimeUnit.MILLISECONDS))
                 .addLast("handler", channelHandler);
     }
 

dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/utils/Constants.java

@@ -35,10 +35,6 @@ public class Constants {
 
     public static final String SLASH = "/";
 
-    public static final int NETTY_SERVER_HEART_BEAT_TIME = 1000 * 60 * 3 + 1000;
-
-    public static final int NETTY_CLIENT_HEART_BEAT_TIME = 1000 * 6;
-
     /**
      * charset
      */

dolphinscheduler-extract/dolphinscheduler-extract-base/src/test/java/org/apache/dolphinscheduler/extract/base/metrics/RpcMetricsTest.java

@@ -0,0 +1,70 @@
+/*
+ * 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.dolphinscheduler.extract.base.metrics;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import org.apache.dolphinscheduler.common.utils.NetUtils;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import io.micrometer.core.instrument.Metrics;
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
+
+class RpcMetricsTest {
+
+    @BeforeEach
+    public void setup() {
+        Metrics.globalRegistry.clear();
+        Metrics.addRegistry(new SimpleMeterRegistry());
+    }
+
+    @Test
+    void testRecordClientSyncRequestException() {
+        assertThat(Metrics.globalRegistry.find("ds.rpc.client.sync.request.exception.count").counter()).isNull();
+
+        String clientHost = NetUtils.getHost();
+        String serverHost = NetUtils.getHost();
+
+        RpcMetrics.recordClientSyncRequestException(
+                new IllegalArgumentException("id is null"), "getById", clientHost, serverHost);
+        RpcMetrics.recordClientSyncRequestException(
+                new IllegalArgumentException("name is null"), "getByName", clientHost, serverHost);
+        RpcMetrics.recordClientSyncRequestException(
+                new IllegalArgumentException("age is null"), "getByAge", clientHost, serverHost);
+        RpcMetrics.recordClientSyncRequestException(new UnsupportedOperationException("update id is not supported"),
+                "updateById", clientHost, serverHost);
+        assertThat(Metrics.globalRegistry.find("ds.rpc.client.sync.request.exception.count").counter()).isNotNull();
+    }
+
+    @Test
+    void testRecordRpcRequestDuration() {
+        assertThat(Metrics.globalRegistry.find("ds.rpc.client.sync.request.duration.time").timer()).isNull();
+
+        String clientHost = NetUtils.getHost();
+        String serverHost = NetUtils.getHost();
+
+        RpcMetrics.recordClientSyncRequestDuration("getById", 100, clientHost, serverHost);
+        RpcMetrics.recordClientSyncRequestDuration("getByName", 200, clientHost, serverHost);
+        RpcMetrics.recordClientSyncRequestDuration("getByAge", 300, clientHost, serverHost);
+        RpcMetrics.recordClientSyncRequestDuration("updateById", 400, clientHost, serverHost);
+        assertThat(Metrics.globalRegistry.find("ds.rpc.client.sync.request.duration.time").timer()).isNotNull();
+    }
+
+}