This is an automated email from the ASF dual-hosted git repository.
zihaoxiang pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
The following commit(s) were added to refs/heads/dev by this push:
new 7a8d910e70 [DSIP-40][APIService] Add LogClient to fetch log (#17165)
7a8d910e70 is described below
commit 7a8d910e705d09c225536b56787f549f5131c88d
Author: [email protected] <[email protected]>
AuthorDate: Tue Jun 10 11:24:40 2025 +0800
[DSIP-40][APIService] Add LogClient to fetch log (#17165)
---
.../api/executor/logging/LocalLogClient.java | 88 ++++++++++++
.../api/executor/logging/LogClientDelegate.java | 112 +++++++++++++++
.../api/executor/logging/RemoteLogClient.java | 54 ++++++++
.../api/service/impl/LoggerServiceImpl.java | 31 +----
.../api/executor/logging/LocalLogClientTest.java | 137 +++++++++++++++++++
.../executor/logging/LogClientDelegateTest.java | 152 +++++++++++++++++++++
.../api/service/LoggerServiceTest.java | 82 +++--------
.../common/service/impl/LogServiceImpl.java | 85 ++++++++++++
...ownloadResponse.java => LogResponseStatus.java} | 22 +--
.../TaskInstanceLogFileDownloadResponse.java | 4 +
.../TaskInstanceLogPageQueryResponse.java | 4 +
.../server/master/rpc/MasterLogServiceImpl.java | 35 +----
.../server/worker/rpc/WorkerLogServiceImpl.java | 34 +----
13 files changed, 675 insertions(+), 165 deletions(-)
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/logging/LocalLogClient.java
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/logging/LocalLogClient.java
new file mode 100644
index 0000000000..ebe8b2e9fe
--- /dev/null
+++
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/logging/LocalLogClient.java
@@ -0,0 +1,88 @@
+/*
+ * 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.api.executor.logging;
+
+import org.apache.dolphinscheduler.dao.entity.TaskInstance;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
+import org.apache.dolphinscheduler.extract.common.ILogService;
+import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogFileDownloadRequest;
+import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogFileDownloadResponse;
+import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogPageQueryRequest;
+import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogPageQueryResponse;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.springframework.stereotype.Component;
+
+@Component
+@Slf4j
+public class LocalLogClient {
+
+ /**
+ * Download the complete log of a task instance.
+ * This method is used to retrieve all log information from the start to
the end of a task instance,
+ * suitable for scenarios where a complete log record is required.
+ *
+ * @param taskInstance The task instance object, containing information
needed to retrieve the log.
+ * @return The complete log file download response of the task instance,
including log content and metadata.
+ */
+ public TaskInstanceLogFileDownloadResponse getWholeLog(TaskInstance
taskInstance) {
+ return getLocalWholeLog(taskInstance);
+ }
+
+ /**
+ * Query a portion of the log of a task instance.
+ * This method is used to query log information of a task instance in a
paginated manner,
+ * suitable for scenarios where the log content is large and needs to be
retrieved in batches.
+ *
+ * @param taskInstance The task instance object, containing information
needed to retrieve the log.
+ * @param skipLineNum The number of lines to skip, indicating from which
line to start reading the log.
+ * @param limit The maximum number of lines to read, indicating the
maximum number of lines to retrieve in this query.
+ * @return The partial log query response, including log content within
the specified range and metadata.
+ */
+ public TaskInstanceLogPageQueryResponse getPartLog(TaskInstance
taskInstance, int skipLineNum, int limit) {
+ return getLocalPartLog(taskInstance, skipLineNum, limit);
+ }
+
+ private TaskInstanceLogFileDownloadResponse getLocalWholeLog(TaskInstance
taskInstance) {
+ TaskInstanceLogFileDownloadRequest request = new
TaskInstanceLogFileDownloadRequest(
+ taskInstance.getId(),
+ taskInstance.getLogPath());
+ return
getProxyLogService(taskInstance).getTaskInstanceWholeLogFileBytes(request);
+ }
+
+ private TaskInstanceLogPageQueryResponse getLocalPartLog(TaskInstance
taskInstance, int skipLineNum,
+ int limit) {
+ TaskInstanceLogPageQueryRequest request =
TaskInstanceLogPageQueryRequest
+ .builder()
+ .taskInstanceId(taskInstance.getId())
+ .taskInstanceLogAbsolutePath(taskInstance.getLogPath())
+ .skipLineNum(skipLineNum)
+ .limit(limit)
+ .build();
+ return
getProxyLogService(taskInstance).pageQueryTaskInstanceLog(request);
+ }
+
+ private ILogService getProxyLogService(TaskInstance taskInstance) {
+ ILogService logService = Clients
+ .withService(ILogService.class)
+ .withHost(taskInstance.getHost());
+ log.debug("Created log service for host: {}", taskInstance.getHost());
+ return logService;
+ }
+}
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/logging/LogClientDelegate.java
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/logging/LogClientDelegate.java
new file mode 100644
index 0000000000..ac5b6ecb2b
--- /dev/null
+++
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/logging/LogClientDelegate.java
@@ -0,0 +1,112 @@
+/*
+ * 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.api.executor.logging;
+
+import org.apache.dolphinscheduler.dao.entity.TaskInstance;
+import
org.apache.dolphinscheduler.extract.common.transportor.LogResponseStatus;
+import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogFileDownloadResponse;
+import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogPageQueryResponse;
+import org.apache.dolphinscheduler.plugin.task.api.utils.TaskTypeUtils;
+import org.apache.dolphinscheduler.registry.api.RegistryClient;
+import org.apache.dolphinscheduler.registry.api.enums.RegistryNodeType;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+public class LogClientDelegate {
+
+ @Autowired
+ private LocalLogClient localLogClient;
+ @Autowired
+ private RemoteLogClient remoteLogClient;
+ @Autowired
+ private RegistryClient registryClient;
+
+ /**
+ * Retrieves a portion of the log string for a given task instance.
+ * This method first attempts to fetch the log from local storage; if
unsuccessful, it tries to obtain the log from remote storage.
+ *
+ * @param taskInstance The task instance object, containing information
needed for log retrieval.
+ * @param skipLineNum The number of log lines to skip from the beginning.
+ * @param limit The maximum number of log lines to retrieve.
+ * @return A string containing the specified portion of the log.
+ */
+ public String getPartLogString(TaskInstance taskInstance, int skipLineNum,
int limit) {
+ checkArgs(taskInstance);
+ if (checkNodeExists(taskInstance)) {
+ TaskInstanceLogPageQueryResponse response =
localLogClient.getPartLog(taskInstance, skipLineNum, limit);
+ if (response.getCode() == LogResponseStatus.SUCCESS) {
+ return response.getLogContent();
+ } else {
+ log.warn("get part log string is not success for task instance
{}; reason :{}",
+ taskInstance.getId(), response.getMessage());
+ return remoteLogClient.getPartLog(taskInstance, skipLineNum,
limit);
+ }
+ } else {
+ return remoteLogClient.getPartLog(taskInstance, skipLineNum,
limit);
+ }
+ }
+
+ /**
+ * Retrieves the complete log content for a given task instance as a byte
array.
+ * This method first attempts to fetch the log from local storage; if
unsuccessful, it tries to obtain the log from remote storage.
+ *
+ * @param taskInstance The task instance object, containing information
needed for log retrieval.
+ * @return A byte array containing the complete log content.
+ */
+ public byte[] getWholeLogBytes(TaskInstance taskInstance) {
+ checkArgs(taskInstance);
+ if (checkNodeExists(taskInstance)) {
+ TaskInstanceLogFileDownloadResponse response =
localLogClient.getWholeLog(taskInstance);
+ if (response.getCode() == LogResponseStatus.SUCCESS) {
+ return response.getLogBytes();
+ } else {
+ log.warn("get whole log bytes is not success for task instance
{}; reason :{}", taskInstance.getId(),
+ response.getMessage());
+ return remoteLogClient.getWholeLog(taskInstance);
+ }
+ } else {
+ return remoteLogClient.getWholeLog(taskInstance);
+ }
+ }
+
+ private static void checkArgs(TaskInstance taskInstance) {
+ if (taskInstance == null) {
+ throw new IllegalArgumentException("canFetchLog task instance is
null");
+ }
+ }
+
+ private boolean checkNodeExists(TaskInstance taskInstance) {
+ RegistryNodeType nodeType;
+ if (TaskTypeUtils.isLogicTask(taskInstance.getTaskType())) {
+ nodeType = RegistryNodeType.MASTER;
+ } else {
+ nodeType = RegistryNodeType.WORKER;
+ }
+ boolean exists =
registryClient.checkNodeExists(taskInstance.getHost(), nodeType);
+ if (!exists) {
+ log.warn("Node {} does not exist for task instance {}",
taskInstance.getHost(), taskInstance.getId());
+ }
+ return exists;
+ }
+
+}
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/logging/RemoteLogClient.java
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/logging/RemoteLogClient.java
new file mode 100644
index 0000000000..1b3542e962
--- /dev/null
+++
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/logging/RemoteLogClient.java
@@ -0,0 +1,54 @@
+/*
+ * 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.api.executor.logging;
+
+import org.apache.dolphinscheduler.common.utils.LogUtils;
+import org.apache.dolphinscheduler.dao.entity.TaskInstance;
+
+import org.springframework.stereotype.Component;
+
+@Component
+public class RemoteLogClient {
+
+ /**
+ * Retrieves the entire log content for a given task instance.
+ * This method is used when it is necessary to obtain all the log
information for a task instance.
+ *
+ * @param taskInstance The task instance object, containing information
such as the task ID and log path.
+ * @return Returns the log content in byte array format.
+ */
+ public byte[] getWholeLog(TaskInstance taskInstance) {
+ return
LogUtils.getFileContentBytesFromRemote(taskInstance.getLogPath());
+ }
+
+ /**
+ * Retrieves part of the log content for a given task instance, based on
the specified line number and the number of lines to read.
+ * This method is used when it is necessary to browse a portion of the log
content, allowing for skipping a certain number of lines and limiting the
number of lines read.
+ *
+ * @param taskInstance The task instance object, containing information
such as the task ID and log path.
+ * @param skipLineNum The number of lines to skip, starting from the
beginning of the log.
+ * @param limit The maximum number of lines to read.
+ * @return Returns the specified part of the log content in string format.
+ */
+ public String getPartLog(TaskInstance taskInstance, int skipLineNum, int
limit) {
+ // todo We can optimize requests by the actual range, reducing disk
usage and network traffic.
+ return LogUtils.rollViewLogLines(
+
LogUtils.readPartFileContentFromRemote(taskInstance.getLogPath(), skipLineNum,
limit));
+ }
+
+}
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/LoggerServiceImpl.java
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/LoggerServiceImpl.java
index 1b8cc02b39..a0b3b35fc1 100644
---
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/LoggerServiceImpl.java
+++
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/LoggerServiceImpl.java
@@ -22,6 +22,7 @@ import static
org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationCon
import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.exceptions.ServiceException;
+import org.apache.dolphinscheduler.api.executor.logging.LogClientDelegate;
import org.apache.dolphinscheduler.api.service.LoggerService;
import org.apache.dolphinscheduler.api.service.ProjectService;
import org.apache.dolphinscheduler.api.utils.Result;
@@ -34,12 +35,6 @@ import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.mapper.ProjectMapper;
import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
-import org.apache.dolphinscheduler.extract.base.client.Clients;
-import org.apache.dolphinscheduler.extract.common.ILogService;
-import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogFileDownloadRequest;
-import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogFileDownloadResponse;
-import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogPageQueryRequest;
-import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogPageQueryResponse;
import org.apache.commons.lang3.StringUtils;
@@ -73,6 +68,9 @@ public class LoggerServiceImpl extends BaseServiceImpl
implements LoggerService
@Autowired
private TaskDefinitionMapper taskDefinitionMapper;
+ @Autowired
+ private LogClientDelegate logClientDelegate;
+
/**
* view log
*
@@ -203,17 +201,7 @@ public class LoggerServiceImpl extends BaseServiceImpl
implements LoggerService
}
try {
- TaskInstanceLogPageQueryRequest request =
TaskInstanceLogPageQueryRequest.builder()
- .taskInstanceId(taskInstance.getId())
- .taskInstanceLogAbsolutePath(logPath)
- .skipLineNum(skipLineNum)
- .limit(limit)
- .build();
- final TaskInstanceLogPageQueryResponse response = Clients
- .withService(ILogService.class)
- .withHost(taskInstance.getHost())
- .pageQueryTaskInstanceLog(request);
- String logContent = response.getLogContent();
+ String logContent =
logClientDelegate.getPartLogString(taskInstance, skipLineNum, limit);
if (logContent != null) {
sb.append(logContent);
}
@@ -241,14 +229,7 @@ public class LoggerServiceImpl extends BaseServiceImpl
implements LoggerService
byte[] logBytes;
try {
- final TaskInstanceLogFileDownloadRequest request = new
TaskInstanceLogFileDownloadRequest(
- taskInstance.getId(),
- logPath);
- final TaskInstanceLogFileDownloadResponse response = Clients
- .withService(ILogService.class)
- .withHost(taskInstance.getHost())
- .getTaskInstanceWholeLogFileBytes(request);
- logBytes = response.getLogBytes();
+ logBytes = logClientDelegate.getWholeLogBytes(taskInstance);
return Bytes.concat(head, logBytes);
} catch (Exception ex) {
log.error("Download TaskInstance: {} Log Error",
taskInstance.getName(), ex);
diff --git
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/executor/logging/LocalLogClientTest.java
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/executor/logging/LocalLogClientTest.java
new file mode 100644
index 0000000000..38ed869007
--- /dev/null
+++
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/executor/logging/LocalLogClientTest.java
@@ -0,0 +1,137 @@
+/*
+ * 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.api.executor.logging;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+import org.apache.dolphinscheduler.api.exceptions.ServiceException;
+import org.apache.dolphinscheduler.dao.entity.TaskInstance;
+import org.apache.dolphinscheduler.extract.base.config.NettyServerConfig;
+import
org.apache.dolphinscheduler.extract.base.server.SpringServerMethodInvokerDiscovery;
+import org.apache.dolphinscheduler.extract.common.ILogService;
+import
org.apache.dolphinscheduler.extract.common.transportor.LogResponseStatus;
+import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogFileDownloadRequest;
+import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogFileDownloadResponse;
+import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogPageQueryRequest;
+import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogPageQueryResponse;
+
+import java.io.IOException;
+import java.net.ServerSocket;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
+
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.LENIENT)
+public class LocalLogClientTest {
+
+ @InjectMocks
+ private LocalLogClient localLogClient;
+
+ private SpringServerMethodInvokerDiscovery
springServerMethodInvokerDiscovery;
+
+ private int nettyServerPort = 18080;
+
+ @BeforeEach
+ public void setUp() {
+ try (ServerSocket s = new ServerSocket(0)) {
+ nettyServerPort = s.getLocalPort();
+ } catch (IOException e) {
+ return;
+ }
+
+ springServerMethodInvokerDiscovery = new
SpringServerMethodInvokerDiscovery(
+
NettyServerConfig.builder().serverName("TestLogServer").listenPort(nettyServerPort).build());
+ springServerMethodInvokerDiscovery.start();
+
springServerMethodInvokerDiscovery.registerServerMethodInvokerProvider(new
ILogService() {
+
+ @Override
+ public TaskInstanceLogFileDownloadResponse
getTaskInstanceWholeLogFileBytes(TaskInstanceLogFileDownloadRequest
taskInstanceLogFileDownloadRequest) {
+ if (taskInstanceLogFileDownloadRequest.getTaskInstanceId() ==
1) {
+ return new TaskInstanceLogFileDownloadResponse(new
byte[0], LogResponseStatus.SUCCESS, "");
+ } else if
(taskInstanceLogFileDownloadRequest.getTaskInstanceId() == 10) {
+ return new TaskInstanceLogFileDownloadResponse("log
content".getBytes(), LogResponseStatus.SUCCESS,
+ "");
+ }
+
+ throw new ServiceException("download error");
+ }
+
+ @Override
+ public TaskInstanceLogPageQueryResponse
pageQueryTaskInstanceLog(TaskInstanceLogPageQueryRequest
taskInstanceLogPageQueryRequest) {
+ if (taskInstanceLogPageQueryRequest.getTaskInstanceId() !=
null) {
+ if (taskInstanceLogPageQueryRequest.getTaskInstanceId() ==
100) {
+ throw new ServiceException("query log error");
+ } else if
(taskInstanceLogPageQueryRequest.getTaskInstanceId() == 10) {
+ return new TaskInstanceLogPageQueryResponse("Partial
log content", LogResponseStatus.SUCCESS,
+ "");
+ }
+ }
+
+ return new TaskInstanceLogPageQueryResponse();
+ }
+
+ @Override
+ public void removeTaskInstanceLog(String
taskInstanceLogAbsolutePath) {
+
+ }
+ });
+ springServerMethodInvokerDiscovery.start();
+ }
+
+ @AfterEach
+ public void tearDown() {
+ if (springServerMethodInvokerDiscovery != null) {
+ springServerMethodInvokerDiscovery.close();
+ }
+ }
+
+ @Test
+ public void testGetWholeLogSuccess() {
+ TaskInstance taskInstance = new TaskInstance();
+ taskInstance.setHost("127.0.0.1:" + nettyServerPort);
+ taskInstance.setId(1);
+ taskInstance.setLogPath("/path/to/log");
+
+ TaskInstanceLogFileDownloadResponse actualResponse =
localLogClient.getWholeLog(taskInstance);
+
+ assertNotNull(actualResponse);
+ assertArrayEquals("".getBytes(), actualResponse.getLogBytes());
+ }
+
+ @Test
+ public void testGetPartLogSuccess() {
+ TaskInstance taskInstance = new TaskInstance();
+ taskInstance.setId(10);
+ taskInstance.setHost("127.0.0.1:" + nettyServerPort);
+ taskInstance.setLogPath("/path/to/log");
+
+ TaskInstanceLogPageQueryResponse actualResponse =
localLogClient.getPartLog(taskInstance, 0, 10);
+
+ assertNotNull(actualResponse);
+ assertEquals("Partial log content", actualResponse.getLogContent());
+ }
+}
diff --git
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/executor/logging/LogClientDelegateTest.java
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/executor/logging/LogClientDelegateTest.java
new file mode 100644
index 0000000000..edf85268a7
--- /dev/null
+++
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/executor/logging/LogClientDelegateTest.java
@@ -0,0 +1,152 @@
+/*
+ * 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.api.executor.logging;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.when;
+
+import org.apache.dolphinscheduler.dao.entity.TaskInstance;
+import
org.apache.dolphinscheduler.extract.common.transportor.LogResponseStatus;
+import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogFileDownloadResponse;
+import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogPageQueryResponse;
+import org.apache.dolphinscheduler.registry.api.RegistryClient;
+import org.apache.dolphinscheduler.registry.api.enums.RegistryNodeType;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+@ExtendWith(MockitoExtension.class)
+public class LogClientDelegateTest {
+
+ @Mock
+ private LocalLogClient localLogClient;
+
+ @Mock
+ private RemoteLogClient remoteLogClient;
+
+ @Mock
+ private RegistryClient registryClient;
+
+ @InjectMocks
+ private LogClientDelegate logClientDelegate;
+
+ @Test
+ public void testGetPartLogStringTaskInstanceNullThrowsException() {
+ assertThrows(IllegalArgumentException.class, () ->
logClientDelegate.getPartLogString(null, 0, 10));
+ }
+
+ @Test
+ public void testGetPartLogStringNodeExistsLocalSuccess() {
+ TaskInstance taskInstance = new TaskInstance();
+ taskInstance.setId(1);
+ taskInstance.setHost("localhost");
+ taskInstance.setTaskType("SHELL");
+ when(registryClient.checkNodeExists(eq(taskInstance.getHost()),
any())).thenReturn(true);
+ when(localLogClient.getPartLog(taskInstance, 0, 10))
+ .thenReturn(new TaskInstanceLogPageQueryResponse("logContent",
LogResponseStatus.SUCCESS, ""));
+ String result = logClientDelegate.getPartLogString(taskInstance, 0,
10);
+ assertEquals("logContent", result);
+ }
+
+ @Test
+ public void testGetPartLogStringNodeExistsLocalFailure() {
+ TaskInstance taskInstance = new TaskInstance();
+ taskInstance.setId(1);
+ taskInstance.setHost("localhost");
+ taskInstance.setTaskType("SHELL");
+
+ when(registryClient.checkNodeExists("localhost",
RegistryNodeType.WORKER)).thenReturn(true);
+ when(localLogClient.getPartLog(taskInstance, 0, 10)).thenReturn(
+ new TaskInstanceLogPageQueryResponse(null,
LogResponseStatus.ERROR, "error"));
+ when(remoteLogClient.getPartLog(taskInstance, 0,
10)).thenReturn("remoteLogContent");
+
+ String result = logClientDelegate.getPartLogString(taskInstance, 0,
10);
+ assertEquals("remoteLogContent", result);
+ }
+
+ @Test
+ public void testGetPartLogStringNodeNotExists() {
+ TaskInstance taskInstance = new TaskInstance();
+ taskInstance.setId(1);
+ taskInstance.setHost("localhost");
+ taskInstance.setTaskType("SHELL");
+
+ when(registryClient.checkNodeExists("localhost",
RegistryNodeType.WORKER)).thenReturn(false);
+ when(remoteLogClient.getPartLog(taskInstance, 0,
10)).thenReturn("remoteLogContent");
+
+ String result = logClientDelegate.getPartLogString(taskInstance, 0,
10);
+ assertEquals("remoteLogContent", result);
+ }
+
+ @Test
+ public void testGetWholeLogBytesTaskInstanceNullThrowsException() {
+ assertThrows(IllegalArgumentException.class, () ->
logClientDelegate.getWholeLogBytes(null));
+ }
+
+ @Test
+ public void testGetWholeLogBytesNodeExistsLocalSuccess() {
+ TaskInstance taskInstance = new TaskInstance();
+ taskInstance.setId(1);
+ taskInstance.setHost("localhost");
+ taskInstance.setTaskType("SWITCH");
+
+ when(registryClient.checkNodeExists("localhost",
RegistryNodeType.MASTER)).thenReturn(true);
+ when(localLogClient.getWholeLog(taskInstance)).thenReturn(
+ new TaskInstanceLogFileDownloadResponse("logBytes".getBytes(),
LogResponseStatus.SUCCESS, null));
+
+ byte[] result = logClientDelegate.getWholeLogBytes(taskInstance);
+ assertArrayEquals("logBytes".getBytes(), result);
+ }
+
+ @Test
+ public void testGetWholeLogBytesNodeExistsLocalFailure() {
+ TaskInstance taskInstance = new TaskInstance();
+ taskInstance.setId(1);
+ taskInstance.setHost("localhost");
+ taskInstance.setTaskType("SWITCH");
+
+ when(registryClient.checkNodeExists("localhost",
RegistryNodeType.MASTER)).thenReturn(true);
+ when(localLogClient.getWholeLog(taskInstance)).thenReturn(
+ new TaskInstanceLogFileDownloadResponse(null,
LogResponseStatus.ERROR, "error"));
+
when(remoteLogClient.getWholeLog(taskInstance)).thenReturn("remoteLogBytes".getBytes());
+
+ byte[] result = logClientDelegate.getWholeLogBytes(taskInstance);
+ assertArrayEquals("remoteLogBytes".getBytes(), result);
+ }
+
+ @Test
+ public void testGetWholeLogBytesNodeNotExists() {
+ TaskInstance taskInstance = new TaskInstance();
+ taskInstance.setId(1);
+ taskInstance.setHost("localhost");
+ taskInstance.setTaskType("SWITCH");
+
+ when(registryClient.checkNodeExists("localhost",
RegistryNodeType.MASTER)).thenReturn(false);
+
when(remoteLogClient.getWholeLog(taskInstance)).thenReturn("remoteLogBytes".getBytes());
+
+ byte[] result = logClientDelegate.getWholeLogBytes(taskInstance);
+ assertArrayEquals("remoteLogBytes".getBytes(), result);
+ }
+}
diff --git
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/LoggerServiceTest.java
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/LoggerServiceTest.java
index 306cf91de2..8788696cbf 100644
---
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/LoggerServiceTest.java
+++
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/LoggerServiceTest.java
@@ -22,6 +22,8 @@ import static
org.apache.dolphinscheduler.api.AssertionsHelper.assertThrowsServi
import static
org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.DOWNLOAD_LOG;
import static
org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.VIEW_LOG;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.when;
@@ -29,6 +31,7 @@ import static org.mockito.Mockito.when;
import org.apache.dolphinscheduler.api.AssertionsHelper;
import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.exceptions.ServiceException;
+import org.apache.dolphinscheduler.api.executor.logging.LogClientDelegate;
import org.apache.dolphinscheduler.api.service.impl.LoggerServiceImpl;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.constants.Constants;
@@ -40,23 +43,12 @@ import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.mapper.ProjectMapper;
import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
-import org.apache.dolphinscheduler.extract.base.config.NettyServerConfig;
-import
org.apache.dolphinscheduler.extract.base.server.SpringServerMethodInvokerDiscovery;
-import org.apache.dolphinscheduler.extract.common.ILogService;
-import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogFileDownloadRequest;
-import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogFileDownloadResponse;
-import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogPageQueryRequest;
-import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogPageQueryResponse;
-
-import java.io.IOException;
-import java.net.ServerSocket;
+
import java.text.MessageFormat;
import java.util.HashMap;
import java.util.Map;
-import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.InjectMocks;
@@ -88,61 +80,10 @@ public class LoggerServiceTest {
@Mock
private TaskDefinitionMapper taskDefinitionMapper;
- private SpringServerMethodInvokerDiscovery
springServerMethodInvokerDiscovery;
-
- private int nettyServerPort = 18080;
-
- @BeforeEach
- public void setUp() {
- try (ServerSocket s = new ServerSocket(0)) {
- nettyServerPort = s.getLocalPort();
- } catch (IOException e) {
- return;
- }
-
- springServerMethodInvokerDiscovery = new
SpringServerMethodInvokerDiscovery(
-
NettyServerConfig.builder().serverName("TestLogServer").listenPort(nettyServerPort).build());
- springServerMethodInvokerDiscovery.start();
-
springServerMethodInvokerDiscovery.registerServerMethodInvokerProvider(new
ILogService() {
-
- @Override
- public TaskInstanceLogFileDownloadResponse
getTaskInstanceWholeLogFileBytes(TaskInstanceLogFileDownloadRequest
taskInstanceLogFileDownloadRequest) {
- if (taskInstanceLogFileDownloadRequest.getTaskInstanceId() ==
1) {
- return new TaskInstanceLogFileDownloadResponse(new
byte[0]);
- } else if
(taskInstanceLogFileDownloadRequest.getTaskInstanceId() == 10) {
- return new TaskInstanceLogFileDownloadResponse("log
content".getBytes());
- }
-
- throw new ServiceException("download error");
- }
-
- @Override
- public TaskInstanceLogPageQueryResponse
pageQueryTaskInstanceLog(TaskInstanceLogPageQueryRequest
taskInstanceLogPageQueryRequest) {
- if (taskInstanceLogPageQueryRequest.getTaskInstanceId() !=
null) {
- if (taskInstanceLogPageQueryRequest.getTaskInstanceId() ==
100) {
- throw new ServiceException("query log error");
- } else if
(taskInstanceLogPageQueryRequest.getTaskInstanceId() == 10) {
- return new TaskInstanceLogPageQueryResponse("log
content");
- }
- }
-
- return new TaskInstanceLogPageQueryResponse();
- }
-
- @Override
- public void removeTaskInstanceLog(String
taskInstanceLogAbsolutePath) {
-
- }
- });
- springServerMethodInvokerDiscovery.start();
- }
+ @Mock
+ private LogClientDelegate logClientDelegate;
- @AfterEach
- public void tearDown() {
- if (springServerMethodInvokerDiscovery != null) {
- springServerMethodInvokerDiscovery.close();
- }
- }
+ private final int nettyServerPort = 18080;
@Test
public void testQueryLog() {
@@ -237,8 +178,10 @@ public class LoggerServiceTest {
() -> loggerService.queryLog(loginUser, 1, 1, 1));
// SUCCESS
+ when(logClientDelegate.getWholeLogBytes(any())).thenReturn(new
byte[0]);
doNothing().when(projectService).checkProjectAndAuthThrowException(loginUser,
taskInstance.getProjectCode(),
DOWNLOAD_LOG);
+ when(logClientDelegate.getWholeLogBytes(any())).thenReturn(new
byte[0]);
byte[] logBytes = loggerService.getLogBytes(loginUser, 1);
Assertions.assertEquals(42, logBytes.length -
String.valueOf(nettyServerPort).length());
}
@@ -278,11 +221,16 @@ public class LoggerServiceTest {
taskDefinition.setProjectCode(1);
taskInstance.setId(10);
when(taskInstanceDao.queryById(10)).thenReturn(taskInstance);
+
+ when(logClientDelegate.getPartLogString(any(), anyInt(),
anyInt())).thenReturn("log content");
+
String result = loggerService.queryLog(loginUser, projectCode, 10, 1,
1);
assertEquals("log content", result);
taskInstance.setId(100);
when(taskInstanceDao.queryById(100)).thenReturn(taskInstance);
+ doThrow(new ServiceException("query log
error")).when(logClientDelegate).getPartLogString(any(), anyInt(),
+ anyInt());
assertThrowsServiceException(Status.QUERY_TASK_INSTANCE_LOG_ERROR,
() -> loggerService.queryLog(loginUser, projectCode, 10, 1,
1));
}
@@ -314,6 +262,7 @@ public class LoggerServiceTest {
when(taskInstanceDao.queryById(1)).thenReturn(taskInstance);
when(taskDefinitionMapper.queryByCode(taskInstance.getTaskCode())).thenReturn(taskDefinition);
+ when(logClientDelegate.getWholeLogBytes(any())).thenReturn(new
byte[0]);
assertDoesNotThrow(() -> loggerService.getLogBytes(loginUser,
projectCode, 1));
taskDefinition.setProjectCode(2L);
@@ -323,6 +272,7 @@ public class LoggerServiceTest {
taskDefinition.setProjectCode(1L);
taskInstance.setId(100);
when(taskInstanceDao.queryById(100)).thenReturn(taskInstance);
+ doThrow(new ServiceException("download
error")).when(logClientDelegate).getWholeLogBytes(any());
assertThrowsServiceException(Status.DOWNLOAD_TASK_INSTANCE_LOG_FILE_ERROR,
() -> loggerService.getLogBytes(loginUser, projectCode, 100));
}
diff --git
a/dolphinscheduler-extract/dolphinscheduler-extract-common/src/main/java/org/apache/dolphinscheduler/extract/common/service/impl/LogServiceImpl.java
b/dolphinscheduler-extract/dolphinscheduler-extract-common/src/main/java/org/apache/dolphinscheduler/extract/common/service/impl/LogServiceImpl.java
new file mode 100644
index 0000000000..bbe2c09d92
--- /dev/null
+++
b/dolphinscheduler-extract/dolphinscheduler-extract-common/src/main/java/org/apache/dolphinscheduler/extract/common/service/impl/LogServiceImpl.java
@@ -0,0 +1,85 @@
+/*
+ * 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.common.service.impl;
+
+import org.apache.dolphinscheduler.common.utils.FileUtils;
+import org.apache.dolphinscheduler.common.utils.LogUtils;
+import org.apache.dolphinscheduler.extract.common.ILogService;
+import
org.apache.dolphinscheduler.extract.common.transportor.LogResponseStatus;
+import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogFileDownloadRequest;
+import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogFileDownloadResponse;
+import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogPageQueryRequest;
+import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogPageQueryResponse;
+
+import org.apache.commons.lang3.exception.ExceptionUtils;
+
+import java.util.List;
+
+public class LogServiceImpl implements ILogService {
+
+ /**
+ * Downloads the entire log file for a task instance.
+ *
+ * @param taskInstanceLogFileDownloadRequest Request object containing the
path to the task instance log file.
+ * @return Response object containing the log file content in byte array
form.
+ */
+ @Override
+ public TaskInstanceLogFileDownloadResponse
getTaskInstanceWholeLogFileBytes(TaskInstanceLogFileDownloadRequest
taskInstanceLogFileDownloadRequest) {
+ final TaskInstanceLogFileDownloadResponse
taskInstanceLogFileDownloadResponse =
+ new TaskInstanceLogFileDownloadResponse();
+ try {
+ byte[] bytes = LogUtils
+
.getFileContentBytesFromLocal(taskInstanceLogFileDownloadRequest.getTaskInstanceLogAbsolutePath());
+ taskInstanceLogFileDownloadResponse.setLogBytes(bytes);
+ } catch (Exception e) {
+
taskInstanceLogFileDownloadResponse.setCode(LogResponseStatus.ERROR);
+
taskInstanceLogFileDownloadResponse.setMessage(ExceptionUtils.getRootCauseMessage(e));
+ }
+ return taskInstanceLogFileDownloadResponse;
+ }
+
+ /**
+ * Performs paginated queries on task instance logs.
+ *
+ * @param taskInstanceLogPageQueryRequest Request object containing the
path to the task instance log file, the number of lines to skip, and the
maximum number of lines to read.
+ * @return Response object containing the log content.
+ */
+ @Override
+ public TaskInstanceLogPageQueryResponse
pageQueryTaskInstanceLog(TaskInstanceLogPageQueryRequest
taskInstanceLogPageQueryRequest) {
+ final TaskInstanceLogPageQueryResponse
taskInstanceLogPageQueryResponse =
+ new TaskInstanceLogPageQueryResponse();
+ List<String> lines;
+ try {
+ lines = LogUtils.readPartFileContentFromLocal(
+
taskInstanceLogPageQueryRequest.getTaskInstanceLogAbsolutePath(),
+ taskInstanceLogPageQueryRequest.getSkipLineNum(),
+ taskInstanceLogPageQueryRequest.getLimit());
+
taskInstanceLogPageQueryResponse.setLogContent(LogUtils.rollViewLogLines(lines));
+ } catch (Exception e) {
+ taskInstanceLogPageQueryResponse.setCode(LogResponseStatus.ERROR);
+
taskInstanceLogPageQueryResponse.setMessage(ExceptionUtils.getMessage(e));
+ }
+ return taskInstanceLogPageQueryResponse;
+ }
+
+ @Override
+ public void removeTaskInstanceLog(String taskInstanceLogAbsolutePath) {
+ FileUtils.deleteFile(taskInstanceLogAbsolutePath);
+ }
+
+}
diff --git
a/dolphinscheduler-extract/dolphinscheduler-extract-common/src/main/java/org/apache/dolphinscheduler/extract/common/transportor/TaskInstanceLogFileDownloadResponse.java
b/dolphinscheduler-extract/dolphinscheduler-extract-common/src/main/java/org/apache/dolphinscheduler/extract/common/transportor/LogResponseStatus.java
similarity index 79%
copy from
dolphinscheduler-extract/dolphinscheduler-extract-common/src/main/java/org/apache/dolphinscheduler/extract/common/transportor/TaskInstanceLogFileDownloadResponse.java
copy to
dolphinscheduler-extract/dolphinscheduler-extract-common/src/main/java/org/apache/dolphinscheduler/extract/common/transportor/LogResponseStatus.java
index ac1e5b0cfd..c82b6c342c 100644
---
a/dolphinscheduler-extract/dolphinscheduler-extract-common/src/main/java/org/apache/dolphinscheduler/extract/common/transportor/TaskInstanceLogFileDownloadResponse.java
+++
b/dolphinscheduler-extract/dolphinscheduler-extract-common/src/main/java/org/apache/dolphinscheduler/extract/common/transportor/LogResponseStatus.java
@@ -17,15 +17,19 @@
package org.apache.dolphinscheduler.extract.common.transportor;
-import lombok.AllArgsConstructor;
-import lombok.Data;
-import lombok.NoArgsConstructor;
+public enum LogResponseStatus {
+ /**
+ * Success status code.
+ */
+ SUCCESS,
-@Data
-@NoArgsConstructor
-@AllArgsConstructor
-public class TaskInstanceLogFileDownloadResponse {
-
- private byte[] logBytes;
+ /**
+ * General error status code.
+ */
+ ERROR,
+ /**
+ * Log file not found status code.
+ */
+ LOG_FILE_NOT_FOUND,
}
diff --git
a/dolphinscheduler-extract/dolphinscheduler-extract-common/src/main/java/org/apache/dolphinscheduler/extract/common/transportor/TaskInstanceLogFileDownloadResponse.java
b/dolphinscheduler-extract/dolphinscheduler-extract-common/src/main/java/org/apache/dolphinscheduler/extract/common/transportor/TaskInstanceLogFileDownloadResponse.java
index ac1e5b0cfd..6cf6dff623 100644
---
a/dolphinscheduler-extract/dolphinscheduler-extract-common/src/main/java/org/apache/dolphinscheduler/extract/common/transportor/TaskInstanceLogFileDownloadResponse.java
+++
b/dolphinscheduler-extract/dolphinscheduler-extract-common/src/main/java/org/apache/dolphinscheduler/extract/common/transportor/TaskInstanceLogFileDownloadResponse.java
@@ -28,4 +28,8 @@ public class TaskInstanceLogFileDownloadResponse {
private byte[] logBytes;
+ private LogResponseStatus code = LogResponseStatus.SUCCESS;
+
+ private String message;
+
}
diff --git
a/dolphinscheduler-extract/dolphinscheduler-extract-common/src/main/java/org/apache/dolphinscheduler/extract/common/transportor/TaskInstanceLogPageQueryResponse.java
b/dolphinscheduler-extract/dolphinscheduler-extract-common/src/main/java/org/apache/dolphinscheduler/extract/common/transportor/TaskInstanceLogPageQueryResponse.java
index ac769d7677..994ceb96b5 100644
---
a/dolphinscheduler-extract/dolphinscheduler-extract-common/src/main/java/org/apache/dolphinscheduler/extract/common/transportor/TaskInstanceLogPageQueryResponse.java
+++
b/dolphinscheduler-extract/dolphinscheduler-extract-common/src/main/java/org/apache/dolphinscheduler/extract/common/transportor/TaskInstanceLogPageQueryResponse.java
@@ -28,4 +28,8 @@ public class TaskInstanceLogPageQueryResponse {
private String logContent;
+ private LogResponseStatus code = LogResponseStatus.SUCCESS;
+
+ private String message;
+
}
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/MasterLogServiceImpl.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/MasterLogServiceImpl.java
index 98e7461f50..5fae40600f 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/MasterLogServiceImpl.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/MasterLogServiceImpl.java
@@ -17,15 +17,8 @@
package org.apache.dolphinscheduler.server.master.rpc;
-import org.apache.dolphinscheduler.common.utils.FileUtils;
-import org.apache.dolphinscheduler.common.utils.LogUtils;
import org.apache.dolphinscheduler.extract.common.ILogService;
-import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogFileDownloadRequest;
-import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogFileDownloadResponse;
-import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogPageQueryRequest;
-import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogPageQueryResponse;
-
-import java.util.List;
+import org.apache.dolphinscheduler.extract.common.service.impl.LogServiceImpl;
import lombok.extern.slf4j.Slf4j;
@@ -33,30 +26,6 @@ import org.springframework.stereotype.Service;
@Slf4j
@Service
-public class MasterLogServiceImpl implements ILogService {
-
- @Override
- public TaskInstanceLogFileDownloadResponse
getTaskInstanceWholeLogFileBytes(TaskInstanceLogFileDownloadRequest
logicTaskInstanceLogFileDownloadRequest) {
- byte[] bytes =
-
LogUtils.getFileContentBytes(logicTaskInstanceLogFileDownloadRequest.getTaskInstanceLogAbsolutePath());
- // todo: if file not exists, return error result
- return new TaskInstanceLogFileDownloadResponse(bytes);
- }
-
- @Override
- public TaskInstanceLogPageQueryResponse
pageQueryTaskInstanceLog(TaskInstanceLogPageQueryRequest
taskInstanceLogPageQueryRequest) {
-
- List<String> lines = LogUtils.readPartFileContent(
-
taskInstanceLogPageQueryRequest.getTaskInstanceLogAbsolutePath(),
- taskInstanceLogPageQueryRequest.getSkipLineNum(),
- taskInstanceLogPageQueryRequest.getLimit());
-
- String logContent = LogUtils.rollViewLogLines(lines);
- return new TaskInstanceLogPageQueryResponse(logContent);
- }
+public class MasterLogServiceImpl extends LogServiceImpl implements
ILogService {
- @Override
- public void removeTaskInstanceLog(String taskInstanceLogAbsolutePath) {
- FileUtils.deleteFile(taskInstanceLogAbsolutePath);
- }
}
diff --git
a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/rpc/WorkerLogServiceImpl.java
b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/rpc/WorkerLogServiceImpl.java
index b1bca6dfe6..68a36c0c6c 100644
---
a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/rpc/WorkerLogServiceImpl.java
+++
b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/rpc/WorkerLogServiceImpl.java
@@ -17,15 +17,8 @@
package org.apache.dolphinscheduler.server.worker.rpc;
-import org.apache.dolphinscheduler.common.utils.FileUtils;
-import org.apache.dolphinscheduler.common.utils.LogUtils;
import org.apache.dolphinscheduler.extract.common.ILogService;
-import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogFileDownloadRequest;
-import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogFileDownloadResponse;
-import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogPageQueryRequest;
-import
org.apache.dolphinscheduler.extract.common.transportor.TaskInstanceLogPageQueryResponse;
-
-import java.util.List;
+import org.apache.dolphinscheduler.extract.common.service.impl.LogServiceImpl;
import lombok.extern.slf4j.Slf4j;
@@ -33,29 +26,6 @@ import org.springframework.stereotype.Service;
@Slf4j
@Service
-public class WorkerLogServiceImpl implements ILogService {
-
- @Override
- public TaskInstanceLogFileDownloadResponse
getTaskInstanceWholeLogFileBytes(TaskInstanceLogFileDownloadRequest
taskInstanceLogFileDownloadRequest) {
- byte[] bytes = LogUtils
-
.getFileContentBytes(taskInstanceLogFileDownloadRequest.getTaskInstanceLogAbsolutePath());
- // todo: if file not exists, return error result
- return new TaskInstanceLogFileDownloadResponse(bytes);
- }
-
- @Override
- public TaskInstanceLogPageQueryResponse
pageQueryTaskInstanceLog(TaskInstanceLogPageQueryRequest
taskInstanceLogPageQueryRequest) {
- List<String> lines = LogUtils.readPartFileContent(
-
taskInstanceLogPageQueryRequest.getTaskInstanceLogAbsolutePath(),
- taskInstanceLogPageQueryRequest.getSkipLineNum(),
- taskInstanceLogPageQueryRequest.getLimit());
-
- String logContent = LogUtils.rollViewLogLines(lines);
- return new TaskInstanceLogPageQueryResponse(logContent);
- }
+public class WorkerLogServiceImpl extends LogServiceImpl implements
ILogService {
- @Override
- public void removeTaskInstanceLog(String taskInstanceLogAbsolutePath) {
- FileUtils.deleteFile(taskInstanceLogAbsolutePath);
- }
}
