From 2880f360eccbf8f3b03c44beceb61c7a7ffabf88 Mon Sep 17 00:00:00 2001 From: Wenjun Ruan Date: Thu, 6 Jun 2024 16:13:43 +0800 Subject: [PATCH] Add RPC metrics --- .../dolphinscheduler-extract-base/pom.xml | 7 ++ .../extract/base/RpcMethod.java | 2 +- .../extract/base/SyncRequestDto.java | 38 ++++++++ .../base/client/NettyClientHandler.java | 2 +- .../base/client/NettyRemotingClient.java | 85 +++++++++++------ .../base/client/SyncClientMethodInvoker.java | 9 +- .../base/config/NettyClientConfig.java | 12 +++ .../base/config/NettyServerConfig.java | 8 ++ .../metrics/ClientSyncDurationMetrics.java | 56 +++++++++++ .../metrics/ClientSyncExceptionMetrics.java | 56 +++++++++++ .../extract/base/metrics/RpcMetrics.java | 94 +++++++++++++++++++ .../base/server/JdkDynamicServerHandler.java | 7 +- .../base/server/NettyRemotingServer.java | 3 +- .../extract/base/utils/Constants.java | 4 - .../extract/base/metrics/RpcMetricsTest.java | 52 ++++++++++ 15 files changed, 394 insertions(+), 41 deletions(-) create mode 100644 dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/SyncRequestDto.java create mode 100644 dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/metrics/ClientSyncDurationMetrics.java create mode 100644 dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/metrics/ClientSyncExceptionMetrics.java create mode 100644 dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/metrics/RpcMetrics.java create mode 100644 dolphinscheduler-extract/dolphinscheduler-extract-base/src/test/java/org/apache/dolphinscheduler/extract/base/metrics/RpcMetricsTest.java diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-base/pom.xml b/dolphinscheduler-extract/dolphinscheduler-extract-base/pom.xml index 9501d55706..0554ea3c3c 100644 --- a/dolphinscheduler-extract/dolphinscheduler-extract-base/pom.xml +++ b/dolphinscheduler-extract/dolphinscheduler-extract-base/pom.xml @@ -47,6 +47,13 @@ dolphinscheduler-common ${project.version} + + + org.apache.dolphinscheduler + dolphinscheduler-meter + ${project.version} + + io.netty netty-all diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/RpcMethod.java b/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/RpcMethod.java index cd1b778c9a..2dcf689ad8 100644 --- a/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/RpcMethod.java +++ b/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; } diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/SyncRequestDto.java b/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/SyncRequestDto.java new file mode 100644 index 0000000000..7df649cd57 --- /dev/null +++ b/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; + +} diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/client/NettyClientHandler.java b/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/client/NettyClientHandler.java index be570eb577..5f50e2441c 100644 --- a/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/client/NettyClientHandler.java +++ b/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); diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/client/NettyRemotingClient.java b/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/client/NettyRemotingClient.java index 8ded68669d..4aea4d6dfe 100644 --- a/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/client/NettyRemotingClient.java +++ b/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) { diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/client/SyncClientMethodInvoker.java b/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/client/SyncClientMethodInvoker.java index 4731a22d0a..ccbdad945b 100644 --- a/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/client/SyncClientMethodInvoker.java +++ b/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()); } diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/config/NettyClientConfig.java b/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/config/NettyClientConfig.java index a41a439b5c..a00ff540f4 100644 --- a/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/config/NettyClientConfig.java +++ b/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; + } diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/config/NettyServerConfig.java b/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/config/NettyServerConfig.java index 9d4a2ee3d2..6a38b08c15 100644 --- a/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/config/NettyServerConfig.java +++ b/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 */ diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/metrics/ClientSyncDurationMetrics.java b/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/metrics/ClientSyncDurationMetrics.java new file mode 100644 index 0000000000..60124363b7 --- /dev/null +++ b/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; + } + +} diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/metrics/ClientSyncExceptionMetrics.java b/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/metrics/ClientSyncExceptionMetrics.java new file mode 100644 index 0000000000..6f91132547 --- /dev/null +++ b/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; + } + +} diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/metrics/RpcMetrics.java b/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/metrics/RpcMetrics.java new file mode 100644 index 0000000000..ade19c8bd9 --- /dev/null +++ b/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 rpcRequestDurationTimer = new ConcurrentHashMap<>(); + + private static final Map 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") + .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") + .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); + } + +} diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/server/JdkDynamicServerHandler.java b/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/server/JdkDynamicServerHandler.java index f57ff0b609..4f9a7034c8 100644 --- a/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/server/JdkDynamicServerHandler.java +++ b/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); } diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/server/NettyRemotingServer.java b/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/server/NettyRemotingServer.java index 9beeaced3d..9ebf802b1e 100644 --- a/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/server/NettyRemotingServer.java +++ b/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); } diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/utils/Constants.java b/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/utils/Constants.java index 76e3872d31..94b92b2539 100644 --- a/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/utils/Constants.java +++ b/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 */ diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-base/src/test/java/org/apache/dolphinscheduler/extract/base/metrics/RpcMetricsTest.java b/dolphinscheduler-extract/dolphinscheduler-extract-base/src/test/java/org/apache/dolphinscheduler/extract/base/metrics/RpcMetricsTest.java new file mode 100644 index 0000000000..211355c0d1 --- /dev/null +++ b/dolphinscheduler-extract/dolphinscheduler-extract-base/src/test/java/org/apache/dolphinscheduler/extract/base/metrics/RpcMetricsTest.java @@ -0,0 +1,52 @@ +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.Test; + +import io.micrometer.core.instrument.Metrics; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; + +class RpcMetricsTest { + + @Test + public void setup() { + Metrics.globalRegistry.clear(); + Metrics.addRegistry(new SimpleMeterRegistry()); + } + + @Test + void testRecordClientSyncRequestException() { + assertThat(Metrics.globalRegistry.find("ds.rpc.client.sync.request.exception").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").counter()).isNotNull(); + } + + @Test + void testRecordRpcRequestDuration() { + assertThat(Metrics.globalRegistry.find("ds.rpc.client.sync.request.duration").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").timer()).isNotNull(); + } + +}