From 2bd65fb2df68847130c7e1ab6616bbfba95783b2 Mon Sep 17 00:00:00 2001 From: Rick Cheng <38122586+rickchengx@users.noreply.github.com> Date: Mon, 13 Feb 2023 16:38:26 +0800 Subject: [PATCH] [Feature][Remote Logging] Add support for writing task logs to OSS (#13332) --- docs/configs/docsdev.js | 8 + docs/docs/en/guide/remote-logging.md | 35 +++++ docs/docs/zh/guide/remote-logging.md | 35 +++++ .../common/constants/Constants.java | 23 +++ .../log/remote/OssRemoteLogHandler.java | 143 ++++++++++++++++++ .../log/remote/RemoteLogHandleThreadPool.java | 44 ++++++ .../common/log/remote/RemoteLogHandler.java | 25 +++ .../log/remote/RemoteLogHandlerFactory.java | 39 +++++ .../common/log/remote/RemoteLogService.java | 48 ++++++ .../common/log/remote/RemoteLogUtils.java | 79 ++++++++++ .../src/main/resources/common.properties | 18 +++ .../log/remote/OssRemoteLogHandlerTest.java | 53 +++++++ .../runner/WorkflowExecuteRunnable.java | 13 ++ .../processor/LoggerRequestProcessor.java | 63 +++++++- .../task/api/AbstractCommandExecutor.java | 11 ++ .../plugin/task/api/TaskExecutionContext.java | 2 + .../plugin/task/api/utils/LogUtils.java | 2 +- .../runner/WorkerTaskExecuteRunnable.java | 17 +++ 18 files changed, 652 insertions(+), 6 deletions(-) create mode 100644 docs/docs/en/guide/remote-logging.md create mode 100644 docs/docs/zh/guide/remote-logging.md create mode 100644 dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/OssRemoteLogHandler.java create mode 100644 dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/RemoteLogHandleThreadPool.java create mode 100644 dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/RemoteLogHandler.java create mode 100644 dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/RemoteLogHandlerFactory.java create mode 100644 dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/RemoteLogService.java create mode 100644 dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/RemoteLogUtils.java create mode 100644 dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/log/remote/OssRemoteLogHandlerTest.java diff --git a/docs/configs/docsdev.js b/docs/configs/docsdev.js index 6909cb8b97..22968f0c3a 100644 --- a/docs/configs/docsdev.js +++ b/docs/configs/docsdev.js @@ -406,6 +406,10 @@ export default { title: 'Data Quality', link: '/en-us/docs/dev/user_doc/guide/data-quality.html', }, + { + title: 'Remote Logging', + link: '/en-us/docs/dev/user_doc/guide/remote-logging.html', + }, { title: 'Upgrade', children: [ @@ -1046,6 +1050,10 @@ export default { title: '数据质量', link: '/zh-cn/docs/dev/user_doc/guide/data-quality.html', }, + { + title: '远程日志存储', + link: '/zh-cn/docs/dev/user_doc/guide/remote-logging.html', + }, { title: '升级', children: [ diff --git a/docs/docs/en/guide/remote-logging.md b/docs/docs/en/guide/remote-logging.md new file mode 100644 index 0000000000..add7a58485 --- /dev/null +++ b/docs/docs/en/guide/remote-logging.md @@ -0,0 +1,35 @@ +# Remote Logging + +Apache DolphinScheduler supports writing task logs to remote storage. When remote logging is enabled, DolphinScheduler will send the task logs to the specified remote storage asynchronously after the task ends. In addition, when the user views or downloads the task log, if the log file does not exist locally, DolphinScheduler will download the corresponding log file from the remote storage to the local file system. + +## Enabling remote logging + +If you deploy DolphinScheduler in `Cluster` or `Pseudo-Cluster` mode, you need to configure `api-server/conf/common.properties`, `master-server/conf/common.properties` and `worker-server/conf/common.properties`. +If you deploy DolphinScheduler in `Standalone` mode, you only need to configure `standalone-server/conf/common.properties` as follows: + +```properties +# Whether to enable remote logging +remote.logging.enable=false +# if remote.logging.enable = true, set the target of remote logging +remote.logging.target=OSS +# if remote.logging.enable = true, set the log base directory +remote.logging.base.dir=logs +# if remote.logging.enable = true, set the number of threads to send logs to remote storage +remote.logging.thread.pool.size=10 +``` + +## Writing task logs to [Aliyun Object Storage Service (OSS)](https://www.aliyun.com/product/oss) + +Configure `common.properties` as follows: + +```properties +# oss access key id, required if you set remote.logging.target=OSS +remote.logging.oss.access.key.id= +# oss access key secret, required if you set remote.logging.target=OSS +remote.logging.oss.access.key.secret= +# oss bucket name, required if you set remote.logging.target=OSS +remote.logging.oss.bucket.name= +# oss endpoint, required if you set remote.logging.target=OSS +remote.logging.oss.endpoint= +``` + diff --git a/docs/docs/zh/guide/remote-logging.md b/docs/docs/zh/guide/remote-logging.md new file mode 100644 index 0000000000..0adde98e7a --- /dev/null +++ b/docs/docs/zh/guide/remote-logging.md @@ -0,0 +1,35 @@ +# 远程日志存储（Remote Logging） + +Apache DolphinScheduler支持将任务日志传输到远端存储上。当配置开启远程日志存储后，DolphinScheduler将在任务结束后，将对应的任务日志异步地发送到指定的远端存储上。此外，用户在查看或下载任务日志时，若本地没有该日志文件，DolphinScheduler将从远端存储上下载对应的日志文件到本地文件系统。 + +## 开启远程日志存储 + +如果您以 `集群` 模式或者 `伪集群` 模式部署DolphinScheduler，您需要对以下路径的文件进行配置：`api-server/conf/common.properties`，`master-server/conf/common.properties`和 `worker-server/conf/common.properties`； +若您以 `单机` 模式部署DolphinScheduler，您只需要配置 `standalone-server/conf/common.properties`，具体配置如下： + +```properties +# 是否开启远程日志存储 +remote.logging.enable=true +# 任务日志写入的远端存储，目前仅支持OSS +remote.logging.target=OSS +# 任务日志在远端存储上的目录 +remote.logging.base.dir=logs +# 设置向远端存储异步发送日志的线程池大小 +remote.logging.thread.pool.size=10 +``` + +## 将任务日志写入[阿里云对象存储（OSS）](https://www.aliyun.com/product/oss) + +配置`common.propertis`如下： + +```properties +# oss access key id, required if you set remote.logging.target=OSS +remote.logging.oss.access.key.id= +# oss access key secret, required if you set remote.logging.target=OSS +remote.logging.oss.access.key.secret= +# oss bucket name, required if you set remote.logging.target=OSS +remote.logging.oss.bucket.name= +# oss endpoint, required if you set remote.logging.target=OSS +remote.logging.oss.endpoint= +``` + diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/constants/Constants.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/constants/Constants.java index a304ab5a34..792746bddb 100644 --- a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/constants/Constants.java +++ b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/constants/Constants.java @@ -806,4 +806,27 @@ public final class Constants { public static final Integer QUERY_ALL_ON_PROJECT = 1; public static final Integer QUERY_ALL_ON_WORKFLOW = 2; public static final Integer QUERY_ALL_ON_TASK = 3; + + /** + * remote logging + */ + public static final String REMOTE_LOGGING_ENABLE = "remote.logging.enable"; + + public static final String REMOTE_LOGGING_TARGET = "remote.logging.target"; + + public static final String REMOTE_LOGGING_BASE_DIR = "remote.logging.base.dir"; + + public static final String REMOTE_LOGGING_THREAD_POOL_SIZE = "remote.logging.thread.pool.size"; + + /** + * remote logging for OSS + */ + + public static final String REMOTE_LOGGING_OSS_ACCESS_KEY_ID = "remote.logging.oss.access.key.id"; + + public static final String REMOTE_LOGGING_OSS_ACCESS_KEY_SECRET = "remote.logging.oss.access.key.secret"; + + public static final String REMOTE_LOGGING_OSS_BUCKET_NAME = "remote.logging.oss.bucket.name"; + + public static final String REMOTE_LOGGING_OSS_ENDPOINT = "remote.logging.oss.endpoint"; } diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/OssRemoteLogHandler.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/OssRemoteLogHandler.java new file mode 100644 index 0000000000..a8d19af98c --- /dev/null +++ b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/OssRemoteLogHandler.java @@ -0,0 +1,143 @@ +/* + * 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.common.log.remote; + +import org.apache.dolphinscheduler.common.constants.Constants; +import org.apache.dolphinscheduler.common.factory.OssClientFactory; +import org.apache.dolphinscheduler.common.model.OssConnection; +import org.apache.dolphinscheduler.common.utils.PropertyUtils; + +import org.apache.commons.lang3.StringUtils; + +import java.io.Closeable; +import java.io.File; +import java.io.IOException; +import java.nio.file.Path; +import java.nio.file.Paths; + +import lombok.extern.slf4j.Slf4j; + +import com.aliyun.oss.OSS; +import com.aliyun.oss.model.Bucket; +import com.aliyun.oss.model.GetObjectRequest; +import com.aliyun.oss.model.PutObjectRequest; + +@Slf4j +public class OssRemoteLogHandler implements RemoteLogHandler, Closeable { + + private static final int OBJECT_NAME_COUNT = 2; + + private OSS ossClient; + + private String bucketName; + + public OssRemoteLogHandler() { + } + + public void init() { + String accessKeyId = readOssAccessKeyId(); + String accessKeySecret = readOssAccessKeySecret(); + String endpoint = readOssEndpoint(); + ossClient = OssClientFactory.buildOssClient(new OssConnection(accessKeyId, accessKeySecret, endpoint)); + + bucketName = readOssBucketName(); + checkBucketNameExists(bucketName); + } + + @Override + public void sendRemoteLog(String logPath) { + String objectName = getObjectNameFromLogPath(logPath); + + try { + log.info("send remote log {} to OSS {}", logPath, objectName); + PutObjectRequest putObjectRequest = new PutObjectRequest(bucketName, objectName, new File(logPath)); + ossClient.putObject(putObjectRequest); + } catch (Exception e) { + log.error("error while sending remote log {} to OSS {}", logPath, objectName, e); + } + } + + @Override + public void getRemoteLog(String logPath) { + String objectName = getObjectNameFromLogPath(logPath); + + try { + log.info("get remote log on OSS {} to {}", objectName, logPath); + ossClient.getObject(new GetObjectRequest(bucketName, objectName), new File(logPath)); + } catch (Exception e) { + log.error("error while getting remote log on OSS {} to {}", objectName, logPath, e); + } + } + + @Override + public void close() throws IOException { + if (ossClient != null) { + ossClient.shutdown(); + } + } + + private String getObjectNameFromLogPath(String logPath) { + Path path = Paths.get(logPath); + int nameCount = path.getNameCount(); + + if (nameCount < OBJECT_NAME_COUNT) { + return Paths.get(readOssBaseDir(), logPath).toString(); + } else { + return Paths.get(readOssBaseDir(), path.subpath(nameCount - OBJECT_NAME_COUNT, nameCount).toString()) + .toString(); + } + } + + private void checkBucketNameExists(String bucketName) { + if (StringUtils.isBlank(bucketName)) { + throw new IllegalArgumentException(Constants.REMOTE_LOGGING_OSS_BUCKET_NAME + " is empty"); + } + + Bucket existsBucket = ossClient.listBuckets() + .stream() + .filter( + bucket -> bucket.getName().equals(bucketName)) + .findFirst() + .orElseThrow(() -> { + return new IllegalArgumentException( + "bucketName: " + bucketName + " does not exist, you need to create them by yourself"); + }); + + log.info("bucketName: {} has been found", existsBucket.getName()); + } + + private String readOssAccessKeyId() { + return PropertyUtils.getString(Constants.REMOTE_LOGGING_OSS_ACCESS_KEY_ID); + } + + private String readOssAccessKeySecret() { + return PropertyUtils.getString(Constants.REMOTE_LOGGING_OSS_ACCESS_KEY_SECRET); + } + + private String readOssEndpoint() { + return PropertyUtils.getString(Constants.REMOTE_LOGGING_OSS_ENDPOINT); + } + + private String readOssBucketName() { + return PropertyUtils.getString(Constants.REMOTE_LOGGING_OSS_BUCKET_NAME); + } + + private String readOssBaseDir() { + return PropertyUtils.getString(Constants.REMOTE_LOGGING_BASE_DIR); + } +} diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/RemoteLogHandleThreadPool.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/RemoteLogHandleThreadPool.java new file mode 100644 index 0000000000..96e743bb3f --- /dev/null +++ b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/RemoteLogHandleThreadPool.java @@ -0,0 +1,44 @@ +/* + * 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.common.log.remote; + +import org.apache.dolphinscheduler.common.constants.Constants; +import org.apache.dolphinscheduler.common.utils.PropertyUtils; + +import java.util.concurrent.Executor; + +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.scheduling.annotation.EnableAsync; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; + +@Configuration +@EnableAsync +public class RemoteLogHandleThreadPool { + + @Bean + public Executor remoteLogHandleExecutor() { + ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); + executor.setCorePoolSize(PropertyUtils.getInt(Constants.REMOTE_LOGGING_THREAD_POOL_SIZE, 10)); + executor.setMaxPoolSize(PropertyUtils.getInt(Constants.REMOTE_LOGGING_THREAD_POOL_SIZE, 10)); + executor.setThreadNamePrefix("remote-logging-"); + executor.initialize(); + + return executor; + } +} diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/RemoteLogHandler.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/RemoteLogHandler.java new file mode 100644 index 0000000000..4626924304 --- /dev/null +++ b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/RemoteLogHandler.java @@ -0,0 +1,25 @@ +/* + * 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.common.log.remote; + +public interface RemoteLogHandler { + + void sendRemoteLog(String logPath); + + void getRemoteLog(String logPath); +} diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/RemoteLogHandlerFactory.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/RemoteLogHandlerFactory.java new file mode 100644 index 0000000000..a132db8d14 --- /dev/null +++ b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/RemoteLogHandlerFactory.java @@ -0,0 +1,39 @@ +/* + * 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.common.log.remote; + +import org.apache.dolphinscheduler.common.constants.Constants; +import org.apache.dolphinscheduler.common.utils.PropertyUtils; + +import lombok.experimental.UtilityClass; + +@UtilityClass +public class RemoteLogHandlerFactory { + + public RemoteLogHandler getRemoteLogHandler() { + if (!RemoteLogUtils.isRemoteLoggingEnable()) { + return null; + } + if (!"OSS".equals(PropertyUtils.getUpperCaseString(Constants.REMOTE_LOGGING_TARGET))) { + return null; + } + OssRemoteLogHandler ossRemoteLogHandler = new OssRemoteLogHandler(); + ossRemoteLogHandler.init(); + return ossRemoteLogHandler; + } +} diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/RemoteLogService.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/RemoteLogService.java new file mode 100644 index 0000000000..4097675eb1 --- /dev/null +++ b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/RemoteLogService.java @@ -0,0 +1,48 @@ +/* + * 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.common.log.remote; + +import org.apache.dolphinscheduler.common.constants.Constants; +import org.apache.dolphinscheduler.common.utils.PropertyUtils; + +import lombok.extern.slf4j.Slf4j; + +import org.springframework.scheduling.annotation.Async; +import org.springframework.stereotype.Component; + +@Component +@Slf4j +public class RemoteLogService { + + @Async("remoteLogHandleExecutor") + public void asyncSendRemoteLog(String logPath) { + if (RemoteLogUtils.isRemoteLoggingEnable()) { + log.info("Start to send log {} to remote target {}", logPath, + PropertyUtils.getString(Constants.REMOTE_LOGGING_TARGET)); + + RemoteLogHandler remoteLogHandler = RemoteLogHandlerFactory.getRemoteLogHandler(); + if (remoteLogHandler == null) { + log.error("remote log handler is null"); + return; + } + remoteLogHandler.sendRemoteLog(logPath); + log.info("Succeed to send log {} to remote target {}", logPath, + PropertyUtils.getString(Constants.REMOTE_LOGGING_TARGET)); + } + } +} diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/RemoteLogUtils.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/RemoteLogUtils.java new file mode 100644 index 0000000000..9d81a1bfd1 --- /dev/null +++ b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/RemoteLogUtils.java @@ -0,0 +1,79 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.dolphinscheduler.common.log.remote; + +import org.apache.dolphinscheduler.common.constants.Constants; +import org.apache.dolphinscheduler.common.utils.PropertyUtils; + +import java.nio.file.Path; +import java.nio.file.Paths; + +import javax.annotation.PostConstruct; + +import lombok.extern.slf4j.Slf4j; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; + +@Component +@Slf4j +public class RemoteLogUtils { + + private static RemoteLogService remoteLogService; + + @Autowired + private RemoteLogService autowiredRemoteLogService; + + @PostConstruct + private void init() { + remoteLogService = autowiredRemoteLogService; + } + + public static void sendRemoteLog(String logPath) { + if (isRemoteLoggingEnable()) { + // send task logs to remote storage asynchronously + remoteLogService.asyncSendRemoteLog(logPath); + } + } + + public static void getRemoteLog(String logPath) { + if (isRemoteLoggingEnable()) { + log.info("Start to get log {} from remote target {}", logPath, + PropertyUtils.getString(Constants.REMOTE_LOGGING_TARGET)); + + mkdirOfLog(logPath); + RemoteLogHandler remoteLogHandler = RemoteLogHandlerFactory.getRemoteLogHandler(); + if (remoteLogHandler == null) { + log.error("remote log handler is null"); + return; + } + remoteLogHandler.getRemoteLog(logPath); + log.info("End get log {} from remote target {}", logPath, + PropertyUtils.getString(Constants.REMOTE_LOGGING_TARGET)); + } + } + + private static void mkdirOfLog(String logPath) { + Path directory = Paths.get(logPath).getParent(); + directory.toFile().mkdirs(); + } + + public static boolean isRemoteLoggingEnable() { + return PropertyUtils.getBoolean(Constants.REMOTE_LOGGING_ENABLE, Boolean.FALSE); + } +} diff --git a/dolphinscheduler-common/src/main/resources/common.properties b/dolphinscheduler-common/src/main/resources/common.properties index c3a196f41d..513fe53559 100644 --- a/dolphinscheduler-common/src/main/resources/common.properties +++ b/dolphinscheduler-common/src/main/resources/common.properties @@ -141,3 +141,21 @@ appId.collect=log # The default env list will be load by Shell task, e.g. /etc/profile,~/.bash_profile shell.env_source_list= + +# Whether to enable remote logging +remote.logging.enable=false +# if remote.logging.enable = true, set the target of remote logging +remote.logging.target=OSS +# if remote.logging.enable = true, set the log base directory +remote.logging.base.dir=logs +# if remote.logging.enable = true, set the number of threads to send logs to remote storage +remote.logging.thread.pool.size=10 +# oss access key id, required if you set remote.logging.target=OSS +remote.logging.oss.access.key.id= +# oss access key secret, required if you set remote.logging.target=OSS +remote.logging.oss.access.key.secret= +# oss bucket name, required if you set remote.logging.target=OSS +remote.logging.oss.bucket.name= +# oss endpoint, required if you set remote.logging.target=OSS +remote.logging.oss.endpoint= + diff --git a/dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/log/remote/OssRemoteLogHandlerTest.java b/dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/log/remote/OssRemoteLogHandlerTest.java new file mode 100644 index 0000000000..726a262938 --- /dev/null +++ b/dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/log/remote/OssRemoteLogHandlerTest.java @@ -0,0 +1,53 @@ +/* + * 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.common.log.remote; + +import org.apache.dolphinscheduler.common.constants.Constants; +import org.apache.dolphinscheduler.common.utils.PropertyUtils; + +import java.lang.reflect.Method; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.MockedStatic; +import org.mockito.Mockito; +import org.mockito.junit.jupiter.MockitoExtension; + +@ExtendWith(MockitoExtension.class) +public class OssRemoteLogHandlerTest { + + @Test + public void testGetObjectNameFromLogPath() throws Exception { + final String logPath = "/path/to/dolphinscheduler/logs/20230116/8245922982496_1-1-3.log"; + final String expectedObjectName = "logs/20230116/8245922982496_1-1-3.log"; + + OssRemoteLogHandler ossRemoteLogHandler = new OssRemoteLogHandler(); + + try (MockedStatic propertyUtilsMockedStatic = Mockito.mockStatic(PropertyUtils.class)) { + propertyUtilsMockedStatic.when(() -> PropertyUtils.getString(Constants.REMOTE_LOGGING_BASE_DIR)) + .thenReturn("logs"); + + Method method = OssRemoteLogHandler.class.getDeclaredMethod("getObjectNameFromLogPath", String.class); + method.setAccessible(true); + String objectName = (String) method.invoke(ossRemoteLogHandler, logPath); + + Assertions.assertEquals(expectedObjectName, objectName); + } + } +} diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnable.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnable.java index 5ebbf8c8ff..3596bcae5a 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnable.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnable.java @@ -42,6 +42,7 @@ import org.apache.dolphinscheduler.common.enums.TaskDependType; import org.apache.dolphinscheduler.common.enums.TaskGroupQueueStatus; import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus; import org.apache.dolphinscheduler.common.graph.DAG; +import org.apache.dolphinscheduler.common.log.remote.RemoteLogUtils; import org.apache.dolphinscheduler.common.model.TaskNodeRelation; import org.apache.dolphinscheduler.common.thread.ThreadUtils; import org.apache.dolphinscheduler.common.utils.DateUtils; @@ -460,6 +461,8 @@ public class WorkflowExecuteRunnable implements Callable { taskInstance.getTaskCode(), taskInstance.getState()); this.updateProcessInstanceState(); + + sendTaskLogOnMasterToRemoteIfNeeded(taskInstance.getLogPath(), taskInstance.getHost()); } catch (Exception ex) { log.error("Task finish failed, get a exception, will remove this taskInstance from completeTaskMap", ex); // remove the task from complete map, so that we can finish in the next time. @@ -2178,4 +2181,14 @@ public class WorkflowExecuteRunnable implements Callable { } + private void sendTaskLogOnMasterToRemoteIfNeeded(String logPath, String host) { + if (RemoteLogUtils.isRemoteLoggingEnable() && isExecutedOnMaster(host)) { + RemoteLogUtils.sendRemoteLog(logPath); + log.info("Master sends task log {} to remote storage asynchronously.", logPath); + } + } + + private boolean isExecutedOnMaster(String host) { + return host.endsWith(masterAddress.split(Constants.COLON)[1]); + } } diff --git a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/processor/LoggerRequestProcessor.java b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/processor/LoggerRequestProcessor.java index 38807bf18a..6259998146 100644 --- a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/processor/LoggerRequestProcessor.java +++ b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/processor/LoggerRequestProcessor.java @@ -20,6 +20,7 @@ package org.apache.dolphinscheduler.remote.processor; import static org.apache.dolphinscheduler.common.constants.Constants.APPID_COLLECT; import static org.apache.dolphinscheduler.common.constants.Constants.DEFAULT_COLLECT_WAY; +import org.apache.dolphinscheduler.common.log.remote.RemoteLogUtils; import org.apache.dolphinscheduler.common.utils.JSONUtils; import org.apache.dolphinscheduler.common.utils.PropertyUtils; import org.apache.dolphinscheduler.plugin.task.api.utils.LogUtils; @@ -81,7 +82,7 @@ public class LoggerRequestProcessor implements NettyRequestProcessor { ViewLogRequestCommand viewLogRequest = JSONUtils.parseObject( command.getBody(), ViewLogRequestCommand.class); String viewLogPath = viewLogRequest.getPath(); - String msg = LogUtils.readWholeFileContent(viewLogPath); + String msg = readWholeFileContent(viewLogPath); ViewLogResponseCommand viewLogResponse = new ViewLogResponseCommand(msg); channel.writeAndFlush(viewLogResponse.convert2Command(command.getOpaque())); break; @@ -156,7 +157,7 @@ public class LoggerRequestProcessor implements NettyRequestProcessor { * @param filePath file path * @return byte array of file */ - private byte[] getFileContentBytes(String filePath) { + private byte[] getFileContentBytesFromLocal(String filePath) { try ( InputStream in = new FileInputStream(filePath); ByteArrayOutputStream bos = new ByteArrayOutputStream()) { @@ -172,6 +173,22 @@ public class LoggerRequestProcessor implements NettyRequestProcessor { return new byte[0]; } + private byte[] getFileContentBytesFromRemote(String filePath) { + RemoteLogUtils.getRemoteLog(filePath); + return getFileContentBytesFromLocal(filePath); + } + + private byte[] getFileContentBytes(String filePath) { + File file = new File(filePath); + if (file.exists()) { + return getFileContentBytesFromLocal(filePath); + } + if (RemoteLogUtils.isRemoteLoggingEnable()) { + return getFileContentBytesFromRemote(filePath); + } + return getFileContentBytesFromLocal(filePath); + } + /** * read part file content，can skip any line and read some lines * @@ -180,9 +197,9 @@ public class LoggerRequestProcessor implements NettyRequestProcessor { * @param limit read lines limit * @return part file content */ - private List readPartFileContent(String filePath, - int skipLine, - int limit) { + private List readPartFileContentFromLocal(String filePath, + int skipLine, + int limit) { File file = new File(filePath); if (file.exists() && file.isFile()) { try (Stream stream = Files.lines(Paths.get(filePath))) { @@ -196,4 +213,40 @@ public class LoggerRequestProcessor implements NettyRequestProcessor { return Collections.emptyList(); } + private List readPartFileContentFromRemote(String filePath, + int skipLine, + int limit) { + RemoteLogUtils.getRemoteLog(filePath); + return readPartFileContentFromLocal(filePath, skipLine, limit); + } + + private List readPartFileContent(String filePath, + int skipLine, + int limit) { + File file = new File(filePath); + if (file.exists()) { + return readPartFileContentFromLocal(filePath, skipLine, limit); + } + if (RemoteLogUtils.isRemoteLoggingEnable()) { + return readPartFileContentFromRemote(filePath, skipLine, limit); + } + return readPartFileContentFromLocal(filePath, skipLine, limit); + } + + private String readWholeFileContentFromRemote(String filePath) { + RemoteLogUtils.getRemoteLog(filePath); + return LogUtils.readWholeFileContentFromLocal(filePath); + } + + private String readWholeFileContent(String filePath) { + File file = new File(filePath); + if (file.exists()) { + return LogUtils.readWholeFileContentFromLocal(filePath); + } + if (RemoteLogUtils.isRemoteLoggingEnable()) { + return readWholeFileContentFromRemote(filePath); + } + return LogUtils.readWholeFileContentFromLocal(filePath); + } + } diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/AbstractCommandExecutor.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/AbstractCommandExecutor.java index 201684ddca..faceab2b3d 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/AbstractCommandExecutor.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/AbstractCommandExecutor.java @@ -21,6 +21,7 @@ import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.EXIT_COD import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.EXIT_CODE_KILL; import static org.apache.dolphinscheduler.plugin.task.api.utils.ProcessUtils.getPidsStr; +import org.apache.dolphinscheduler.common.log.remote.RemoteLogUtils; import org.apache.dolphinscheduler.common.utils.PropertyUtils; import org.apache.dolphinscheduler.plugin.task.api.model.TaskResponse; import org.apache.dolphinscheduler.plugin.task.api.utils.AbstractCommandExecutorConstants; @@ -100,6 +101,11 @@ public abstract class AbstractCommandExecutor { this.taskRequest = taskRequest; this.logger = logger; this.logBuffer = new LinkedBlockingQueue<>(); + + if (this.taskRequest != null) { + // set logBufferEnable=true if the task uses logHandler and logBuffer to buffer log messages + this.taskRequest.setLogBufferEnable(true); + } } public AbstractCommandExecutor(LinkedBlockingQueue logBuffer) { @@ -335,6 +341,11 @@ public abstract class AbstractCommandExecutor { logBuffer.clear(); } logHandler.accept(markerLog); + + if (RemoteLogUtils.isRemoteLoggingEnable()) { + RemoteLogUtils.sendRemoteLog(taskRequest.getLogPath()); + logger.info("Log handler sends task log {} to remote storage asynchronously.", taskRequest.getLogPath()); + } } /** diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java index 4e02f954ed..2a676755fc 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java @@ -268,4 +268,6 @@ public class TaskExecutionContext implements Serializable { * test flag */ private int testFlag; + + private boolean logBufferEnable; } diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/utils/LogUtils.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/utils/LogUtils.java index 16849d3837..9bff3b66c1 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/utils/LogUtils.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/utils/LogUtils.java @@ -164,7 +164,7 @@ public class LogUtils { } } - public static String readWholeFileContent(String filePath) { + public static String readWholeFileContentFromLocal(String filePath) { String line; StringBuilder sb = new StringBuilder(); try (BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream(filePath)))) { diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerTaskExecuteRunnable.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerTaskExecuteRunnable.java index 631f7298d3..103e253800 100644 --- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerTaskExecuteRunnable.java +++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerTaskExecuteRunnable.java @@ -23,6 +23,7 @@ import static org.apache.dolphinscheduler.common.constants.Constants.DRY_RUN_FLA import static org.apache.dolphinscheduler.common.constants.Constants.SINGLE_SLASH; import org.apache.dolphinscheduler.common.enums.WarningType; +import org.apache.dolphinscheduler.common.log.remote.RemoteLogUtils; import org.apache.dolphinscheduler.common.utils.DateUtils; import org.apache.dolphinscheduler.common.utils.JSONUtils; import org.apache.dolphinscheduler.common.utils.PropertyUtils; @@ -120,6 +121,8 @@ public abstract class WorkerTaskExecuteRunnable implements Runnable { TaskExecutionContextCacheManager.removeByTaskInstanceId(taskExecutionContext.getTaskInstanceId()); log.info("Remove the current task execute context from worker cache"); clearTaskExecPathIfNeeded(); + + sendTaskLogOnWorkerToRemoteIfNeeded(); } protected void afterThrowing(Throwable throwable) throws TaskException { @@ -131,6 +134,8 @@ public abstract class WorkerTaskExecuteRunnable implements Runnable { log.info( "Get a exception when execute the task, will send the task execute result to master, the current task execute result is {}", TaskExecutionStatus.FAILURE); + + sendTaskLogOnWorkerToRemoteIfNeeded(); } public void cancelTask() { @@ -279,6 +284,18 @@ public abstract class WorkerTaskExecuteRunnable implements Runnable { taskExecutionContext.getCurrentExecutionStatus()); } + protected void sendTaskLogOnWorkerToRemoteIfNeeded() { + if (taskExecutionContext.isLogBufferEnable()) { + return; + } + + if (RemoteLogUtils.isRemoteLoggingEnable()) { + RemoteLogUtils.sendRemoteLog(taskExecutionContext.getLogPath()); + log.info("Worker sends task log {} to remote storage asynchronously.", + taskExecutionContext.getLogPath()); + } + } + protected void clearTaskExecPathIfNeeded() { String execLocalPath = taskExecutionContext.getExecutePath(); if (!CommonUtils.isDevelopMode()) {