You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dolphinscheduler.apache.org by zi...@apache.org on 2025/06/10 03:24:45 UTC
(dolphinscheduler) branch dev updated: [DSIP-40][APIService] Add LogClient to fetch log (#17165)
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: luxl@chinatelecom.cn <li...@126.com>
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);
- }
}