diff --git a/docs/docs/en/guide/metrics/metrics.md b/docs/docs/en/guide/metrics/metrics.md index 3470340465..9ba7cc73e7 100644 --- a/docs/docs/en/guide/metrics/metrics.md +++ b/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 diff --git a/docs/docs/zh/guide/metrics/metrics.md b/docs/docs/zh/guide/metrics/metrics.md index aa620b873e..4865a14222 100644 --- a/docs/docs/zh/guide/metrics/metrics.md +++ b/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过载次数 diff --git a/dolphinscheduler-e2e/README.md b/dolphinscheduler-e2e/README.md index f2fbcead77..eb332de90e 100644 --- a/dolphinscheduler-e2e/README.md +++ b/dolphinscheduler-e2e/README.md @@ -97,3 +97,27 @@ class TenantE2ETest { - For UI tests, it's common that the pages might need some time to load, or the operations might need some time to complete, we can use `await().untilAsserted(() -> {})` to wait for the assertions. +## Local development + +### Mac M1 +Add VM options to the test configuration in IntelliJ IDEA: +``` +# In this mode you need to install docker desktop for mac and run it with locally +-Dm1_chip=true +``` + +### Running locally(without Docker) +``` +# In this mode you need to start frontend and backend services locally +-Dlocal=true +``` + +### Running locally(with Docker) +``` +# In this mode you only need to install docker locally +``` + +- To run the tests locally, you need to have the DolphinScheduler running locally. You should add `dolphinscheduler-e2e/pom.xml` to the maven project + Since it does not participate in project compilation, it is not in the main project. +- Running run test class `org.apache.dolphinscheduler.e2e.cases.UserE2ETest` in the IDE. After execution, the test video will be saved as mp4 in a local temporary directory. Such as + `/var/folders/hf/123/T/record-3123/PASSED-[engine:junit-jupiter]/[class:org.apache.dolphinscheduler.e2e.cases.UserE2ETest]-20240606-152333.mp4` diff --git a/dolphinscheduler-e2e/dolphinscheduler-e2e-core/src/main/java/org/apache/dolphinscheduler/e2e/core/DolphinSchedulerExtension.java b/dolphinscheduler-e2e/dolphinscheduler-e2e-core/src/main/java/org/apache/dolphinscheduler/e2e/core/DolphinSchedulerExtension.java index 21c740952c..dcf22a2d6d 100644 --- a/dolphinscheduler-e2e/dolphinscheduler-e2e-core/src/main/java/org/apache/dolphinscheduler/e2e/core/DolphinSchedulerExtension.java +++ b/dolphinscheduler-e2e/dolphinscheduler-e2e-core/src/main/java/org/apache/dolphinscheduler/e2e/core/DolphinSchedulerExtension.java @@ -134,8 +134,8 @@ final class DolphinSchedulerExtension implements BeforeAllCallback, AfterAllCall private void setBrowserContainerByOsName() { DockerImageName imageName; - if (LOCAL_MODE && M1_CHIP_FLAG) { - imageName = DockerImageName.parse("seleniarm/standalone-chromium:4.1.2-20220227") + if (M1_CHIP_FLAG) { + imageName = DockerImageName.parse("seleniarm/standalone-chromium:124.0-chromedriver-124.0") .asCompatibleSubstituteFor("selenium/standalone-chrome"); browser = new BrowserWebDriverContainer<>(imageName) @@ -143,6 +143,7 @@ final class DolphinSchedulerExtension implements BeforeAllCallback, AfterAllCall .withCreateContainerCmdModifier(cmd -> cmd.withUser("root")) .withFileSystemBind(Constants.HOST_CHROME_DOWNLOAD_PATH.toFile().getAbsolutePath(), Constants.SELENIUM_CONTAINER_CHROME_DOWNLOAD_PATH) + .withRecordingMode(RECORD_ALL, record.toFile(), MP4) .withStartupTimeout(Duration.ofSeconds(300)); } else { browser = new BrowserWebDriverContainer<>() @@ -203,7 +204,7 @@ final class DolphinSchedulerExtension implements BeforeAllCallback, AfterAllCall .map(URL::getPath) .map(File::new) .collect(Collectors.toList()); - + ComposeContainer compose = new ComposeContainer(files) .withPull(true) .withTailChildContainers(true) @@ -213,7 +214,7 @@ final class DolphinSchedulerExtension implements BeforeAllCallback, AfterAllCall DOCKER_PORT, Wait.forListeningPort().withStartupTimeout(Duration.ofSeconds(300))) .withLogConsumer(serviceName, outputFrame -> LOGGER.info(outputFrame.getUtf8String())) .waitingFor(serviceName, Wait.forHealthcheck().withStartupTimeout(Duration.ofSeconds(300))); - + return compose; } diff --git a/dolphinscheduler-e2e/pom.xml b/dolphinscheduler-e2e/pom.xml index c7ce90b1a6..6731240486 100644 --- a/dolphinscheduler-e2e/pom.xml +++ b/dolphinscheduler-e2e/pom.xml @@ -36,7 +36,7 @@ UTF-8 5.8.1 - 4.6.0 + 4.13.0 1.18.20 3.20.2 1.5.30 @@ -119,7 +119,7 @@ org.testcontainers testcontainers-bom - 1.19.3 + 1.19.8 import pom 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..c2dbbfefa6 --- /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.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); + } + +} 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..352e218206 --- /dev/null +++ b/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(); + } + +}