Browse Source

[Feature][Remote Logging] Add support for writing task logs to OSS (#13332)

3.2.0-release
Rick Cheng 2 years ago committed by GitHub
parent
commit
2bd65fb2df
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 8
      docs/configs/docsdev.js
  2. 35
      docs/docs/en/guide/remote-logging.md
  3. 35
      docs/docs/zh/guide/remote-logging.md
  4. 23
      dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/constants/Constants.java
  5. 143
      dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/OssRemoteLogHandler.java
  6. 44
      dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/RemoteLogHandleThreadPool.java
  7. 25
      dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/RemoteLogHandler.java
  8. 39
      dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/RemoteLogHandlerFactory.java
  9. 48
      dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/RemoteLogService.java
  10. 79
      dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/log/remote/RemoteLogUtils.java
  11. 18
      dolphinscheduler-common/src/main/resources/common.properties
  12. 53
      dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/log/remote/OssRemoteLogHandlerTest.java
  13. 13
      dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnable.java
  14. 59
      dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/processor/LoggerRequestProcessor.java
  15. 11
      dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/AbstractCommandExecutor.java
  16. 2
      dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java
  17. 2
      dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/utils/LogUtils.java
  18. 17
      dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/runner/WorkerTaskExecuteRunnable.java

8
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: [

35
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=<access.key.id>
# oss access key secret, required if you set remote.logging.target=OSS
remote.logging.oss.access.key.secret=<access.key.secret>
# oss bucket name, required if you set remote.logging.target=OSS
remote.logging.oss.bucket.name=<bucket.name>
# oss endpoint, required if you set remote.logging.target=OSS
remote.logging.oss.endpoint=<endpoint>
```

35
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=<access.key.id>
# oss access key secret, required if you set remote.logging.target=OSS
remote.logging.oss.access.key.secret=<access.key.secret>
# oss bucket name, required if you set remote.logging.target=OSS
remote.logging.oss.bucket.name=<bucket.name>
# oss endpoint, required if you set remote.logging.target=OSS
remote.logging.oss.endpoint=<endpoint>
```

23
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";
}

143
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);
}
}

44
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;
}
}

25
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);
}

39
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;
}
}

48
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));
}
}
}

79
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);
}
}

18
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=<access.key.id>
# oss access key secret, required if you set remote.logging.target=OSS
remote.logging.oss.access.key.secret=<access.key.secret>
# oss bucket name, required if you set remote.logging.target=OSS
remote.logging.oss.bucket.name=<bucket.name>
# oss endpoint, required if you set remote.logging.target=OSS
remote.logging.oss.endpoint=<endpoint>

53
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<PropertyUtils> 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);
}
}
}

13
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<WorkflowSubmitStatue> {
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<WorkflowSubmitStatue> {
}
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]);
}
}

59
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 contentcan skip any line and read some lines
*
@ -180,7 +197,7 @@ public class LoggerRequestProcessor implements NettyRequestProcessor {
* @param limit read lines limit
* @return part file content
*/
private List<String> readPartFileContent(String filePath,
private List<String> readPartFileContentFromLocal(String filePath,
int skipLine,
int limit) {
File file = new File(filePath);
@ -196,4 +213,40 @@ public class LoggerRequestProcessor implements NettyRequestProcessor {
return Collections.emptyList();
}
private List<String> readPartFileContentFromRemote(String filePath,
int skipLine,
int limit) {
RemoteLogUtils.getRemoteLog(filePath);
return readPartFileContentFromLocal(filePath, skipLine, limit);
}
private List<String> 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);
}
}

11
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<String> 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());
}
}
/**

2
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;
}

2
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)))) {

17
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()) {

Loading…
Cancel
Save