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 643457661f [Feature-18136][API] Support view the running task/workflow
of a active master/worker at UI monitor page (#18138)
643457661f is described below
commit 643457661fa2146b7566c5ca394dc4c59c94319f
Author: Wenjun Ruan <[email protected]>
AuthorDate: Sat Apr 11 10:26:13 2026 +0800
[Feature-18136][API] Support view the running task/workflow of a active
master/worker at UI monitor page (#18138)
---
.../api/controller/MonitorController.java | 34 +++++
.../api/service/MonitorService.java | 6 +
.../api/service/impl/MonitorServiceImpl.java | 44 +++++++
.../extract/base/serialize/JsonSerializer.java | 2 +
.../master/IWorkflowExecutorQueryClient.java | 24 ++--
.../extract/master/dto/TaskInstanceExecuteDto.java | 112 -----------------
.../extract/master/dto/WorkflowExecuteDto.java | 137 --------------------
.../extract/master/dto/WorkflowExecutorDTO.java | 44 ++++---
.../transportor/WorkflowExecutorQueryRequest.java | 15 +--
.../transportor/WorkflowExecutorQueryResponse.java | 47 +++----
.../extract/worker/ITaskExecutorQueryClient.java | 24 ++--
.../transportor/TaskExecutorQueryRequest.java | 15 +--
.../transportor/TaskExecutorQueryResponse.java | 47 +++----
.../server/master/engine/WorkflowEngine.java | 26 ++++
.../engine/executor/LogicTaskEngineDelegator.java | 7 ++
.../task/runnable/TaskExecutionContextBuilder.java | 2 +
.../master/rpc/LogicTaskExecutorQueryClient.java | 56 +++++++++
.../master/rpc/WorkflowExecutorQueryClient.java | 52 ++++++++
.../task/executor/ITaskEngine.java | 10 ++
.../dolphinscheduler/task/executor/TaskEngine.java | 24 ++++
.../task/executor/dto/TaskExecutorDTO.java | 42 ++++---
.../plugin/task/api/TaskExecutionContext.java | 6 +-
dolphinscheduler-ui/src/locales/en_US/monitor.ts | 11 ++
dolphinscheduler-ui/src/locales/zh_CN/monitor.ts | 11 ++
.../src/service/modules/monitor/index.ts | 16 +++
.../src/service/modules/monitor/types.ts | 24 +++-
.../src/views/monitor/servers/master/index.tsx | 81 +++++++++++-
.../servers/master/running-workflows-modal.tsx | 139 +++++++++++++++++++++
.../src/views/monitor/servers/worker/index.tsx | 51 +++++++-
.../monitor/servers/worker/running-tasks-modal.tsx | 135 ++++++++++++++++++++
.../executor/PhysicalTaskEngineDelegator.java | 6 +
.../rpc/PhysicalTaskExecutorQueryClient.java | 56 +++++++++
32 files changed, 924 insertions(+), 382 deletions(-)
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/MonitorController.java
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/MonitorController.java
index 3ae1c78a20..c1f4b3124c 100644
---
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/MonitorController.java
+++
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/MonitorController.java
@@ -17,6 +17,7 @@
package org.apache.dolphinscheduler.api.controller;
+import static
org.apache.dolphinscheduler.api.enums.Status.INTERNAL_SERVER_ERROR_ARGS;
import static org.apache.dolphinscheduler.api.enums.Status.LIST_MASTERS_ERROR;
import static
org.apache.dolphinscheduler.api.enums.Status.QUERY_DATABASE_STATE_ERROR;
@@ -27,7 +28,9 @@ import org.apache.dolphinscheduler.common.constants.Constants;
import org.apache.dolphinscheduler.common.model.Server;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.plugin.api.monitor.DatabaseMetrics;
+import org.apache.dolphinscheduler.extract.master.dto.WorkflowExecutorDTO;
import org.apache.dolphinscheduler.registry.api.enums.RegistryNodeType;
+import org.apache.dolphinscheduler.task.executor.dto.TaskExecutorDTO;
import java.util.List;
@@ -37,6 +40,7 @@ import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestAttribute;
import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseStatus;
import org.springframework.web.bind.annotation.RestController;
@@ -81,4 +85,34 @@ public class MonitorController extends BaseController {
return Result.success(databaseMetrics);
}
+ /**
+ * query running workflow instances on a specific master
+ *
+ * @param masterAddress master address in host:port format
+ * @return running workflow instance list
+ */
+ @GetMapping(value = "/masters/workflow-executors")
+ @ResponseStatus(HttpStatus.OK)
+ @ApiException(INTERNAL_SERVER_ERROR_ARGS)
+ public Result<List<WorkflowExecutorDTO>>
queryWorkflowExecutors(@Parameter(hidden = true) @RequestAttribute(value =
Constants.SESSION_USER) User loginUser,
+
@RequestParam("masterAddress") String masterAddress) {
+ List<WorkflowExecutorDTO> workflows =
monitorService.queryWorkflowExecutors(loginUser, masterAddress);
+ return Result.success(workflows);
+ }
+
+ /**
+ * query running task instances on a specific worker
+ *
+ * @param serverAddress worker address in host:port format
+ * @return running task instance list
+ */
+ @GetMapping(value = "/workers/task-executors")
+ @ResponseStatus(HttpStatus.OK)
+ @ApiException(INTERNAL_SERVER_ERROR_ARGS)
+ public Result<List<TaskExecutorDTO>> queryTaskExecutors(@Parameter(hidden
= true) @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
+
@RequestParam("serverAddress") String serverAddress) {
+ List<TaskExecutorDTO> tasks =
monitorService.queryTaskExecutors(loginUser, serverAddress);
+ return Result.success(tasks);
+ }
+
}
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/MonitorService.java
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/MonitorService.java
index 7c0e2d0bf9..33b8db6a16 100644
---
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/MonitorService.java
+++
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/MonitorService.java
@@ -20,7 +20,9 @@ package org.apache.dolphinscheduler.api.service;
import org.apache.dolphinscheduler.common.model.Server;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.plugin.api.monitor.DatabaseMetrics;
+import org.apache.dolphinscheduler.extract.master.dto.WorkflowExecutorDTO;
import org.apache.dolphinscheduler.registry.api.enums.RegistryNodeType;
+import org.apache.dolphinscheduler.task.executor.dto.TaskExecutorDTO;
import java.util.List;
@@ -41,4 +43,8 @@ public interface MonitorService {
* @return server information list
*/
List<Server> listServer(RegistryNodeType nodeType);
+
+ List<WorkflowExecutorDTO> queryWorkflowExecutors(User loginUser, String
masterAddress);
+
+ List<TaskExecutorDTO> queryTaskExecutors(User loginUser, String
serverAddress);
}
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/MonitorServiceImpl.java
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/MonitorServiceImpl.java
index 79912e9398..9ca024ef96 100644
---
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/MonitorServiceImpl.java
+++
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/MonitorServiceImpl.java
@@ -17,13 +17,25 @@
package org.apache.dolphinscheduler.api.service.impl;
+import org.apache.dolphinscheduler.api.enums.Status;
+import org.apache.dolphinscheduler.api.exceptions.ServiceException;
import org.apache.dolphinscheduler.api.service.MonitorService;
+import org.apache.dolphinscheduler.common.enums.UserType;
import org.apache.dolphinscheduler.common.model.Server;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.plugin.api.monitor.DatabaseMetrics;
import org.apache.dolphinscheduler.dao.plugin.api.monitor.DatabaseMonitor;
+import org.apache.dolphinscheduler.extract.base.client.Clients;
+import org.apache.dolphinscheduler.extract.master.IWorkflowExecutorQueryClient;
+import org.apache.dolphinscheduler.extract.master.dto.WorkflowExecutorDTO;
+import
org.apache.dolphinscheduler.extract.master.transportor.WorkflowExecutorQueryRequest;
+import
org.apache.dolphinscheduler.extract.master.transportor.WorkflowExecutorQueryResponse;
+import org.apache.dolphinscheduler.extract.worker.ITaskExecutorQueryClient;
+import
org.apache.dolphinscheduler.extract.worker.transportor.TaskExecutorQueryRequest;
+import
org.apache.dolphinscheduler.extract.worker.transportor.TaskExecutorQueryResponse;
import org.apache.dolphinscheduler.registry.api.RegistryClient;
import org.apache.dolphinscheduler.registry.api.enums.RegistryNodeType;
+import org.apache.dolphinscheduler.task.executor.dto.TaskExecutorDTO;
import java.util.List;
@@ -59,4 +71,36 @@ public class MonitorServiceImpl extends BaseServiceImpl
implements MonitorServic
public List<Server> listServer(RegistryNodeType nodeType) {
return registryClient.getServerList(nodeType);
}
+
+ @Override
+ public List<WorkflowExecutorDTO> queryWorkflowExecutors(User loginUser,
String masterAddress) {
+ if (!loginUser.getUserType().equals(UserType.ADMIN_USER)) {
+ throw new ServiceException(Status.NO_CURRENT_OPERATING_PERMISSION);
+ }
+
+ WorkflowExecutorQueryResponse response = Clients
+ .withService(IWorkflowExecutorQueryClient.class)
+ .withHost(masterAddress)
+ .queryWorkflowExecutors(new WorkflowExecutorQueryRequest());
+ if (!response.isSuccess()) {
+ throw new ServiceException(response.getMessage());
+ }
+ return response.getWorkflowExecutors();
+ }
+
+ @Override
+ public List<TaskExecutorDTO> queryTaskExecutors(User loginUser, String
serverAddress) {
+ if (!loginUser.getUserType().equals(UserType.ADMIN_USER)) {
+ throw new ServiceException(Status.NO_CURRENT_OPERATING_PERMISSION);
+ }
+ TaskExecutorQueryResponse response = Clients
+ .withService(ITaskExecutorQueryClient.class)
+ .withHost(serverAddress)
+ .queryTaskInstances(new TaskExecutorQueryRequest());
+ if (!response.isSuccess()) {
+ throw new ServiceException(response.getMessage());
+ }
+ return response.getTaskExecutors();
+ }
+
}
diff --git
a/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/serialize/JsonSerializer.java
b/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/serialize/JsonSerializer.java
index 6c5ff034a8..d9245969dc 100644
---
a/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/serialize/JsonSerializer.java
+++
b/dolphinscheduler-extract/dolphinscheduler-extract-base/src/main/java/org/apache/dolphinscheduler/extract/base/serialize/JsonSerializer.java
@@ -21,6 +21,7 @@ import static
com.fasterxml.jackson.databind.DeserializationFeature.ACCEPT_EMPTY
import static
com.fasterxml.jackson.databind.DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES;
import static
com.fasterxml.jackson.databind.DeserializationFeature.READ_UNKNOWN_ENUM_VALUES_AS_NULL;
import static
com.fasterxml.jackson.databind.MapperFeature.REQUIRE_SETTERS_FOR_GETTERS;
+import static
com.fasterxml.jackson.databind.SerializationFeature.FAIL_ON_EMPTY_BEANS;
import static
org.apache.dolphinscheduler.common.constants.DateConstants.YYYY_MM_DD_HH_MM_SS;
import org.apache.dolphinscheduler.common.constants.SystemConstants;
@@ -46,6 +47,7 @@ public class JsonSerializer {
.configure(ACCEPT_EMPTY_ARRAY_AS_NULL_OBJECT, true)
.configure(READ_UNKNOWN_ENUM_VALUES_AS_NULL, true)
.configure(REQUIRE_SETTERS_FOR_GETTERS, true)
+ .configure(FAIL_ON_EMPTY_BEANS, false)
.addModule(new SimpleModule()
.addSerializer(LocalDateTime.class, new
JSONUtils.LocalDateTimeSerializer())
.addDeserializer(LocalDateTime.class, new
JSONUtils.LocalDateTimeDeserializer()))
diff --git a/dolphinscheduler-ui/src/service/modules/monitor/index.ts
b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/IWorkflowExecutorQueryClient.java
similarity index 60%
copy from dolphinscheduler-ui/src/service/modules/monitor/index.ts
copy to
dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/IWorkflowExecutorQueryClient.java
index a1ad1b3415..bd4a0f45c2 100644
--- a/dolphinscheduler-ui/src/service/modules/monitor/index.ts
+++
b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/IWorkflowExecutorQueryClient.java
@@ -15,19 +15,17 @@
* limitations under the License.
*/
-import { axios } from '@/service/service'
-import type { ServerNodeType } from './types'
+package org.apache.dolphinscheduler.extract.master;
-export function queryDatabaseState(): any {
- return axios({
- url: '/monitor/databases',
- method: 'get'
- })
-}
+import org.apache.dolphinscheduler.extract.base.RpcMethod;
+import org.apache.dolphinscheduler.extract.base.RpcService;
+import
org.apache.dolphinscheduler.extract.master.transportor.WorkflowExecutorQueryRequest;
+import
org.apache.dolphinscheduler.extract.master.transportor.WorkflowExecutorQueryResponse;
+
+@RpcService
+public interface IWorkflowExecutorQueryClient {
+
+ @RpcMethod
+ WorkflowExecutorQueryResponse
queryWorkflowExecutors(WorkflowExecutorQueryRequest request);
-export function listMonitorServerNode(nodeType: ServerNodeType): any {
- return axios({
- url: `/monitor/${nodeType}`,
- method: 'get'
- })
}
diff --git
a/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/dto/TaskInstanceExecuteDto.java
b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/dto/TaskInstanceExecuteDto.java
deleted file mode 100644
index fa24a151eb..0000000000
---
a/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/dto/TaskInstanceExecuteDto.java
+++ /dev/null
@@ -1,112 +0,0 @@
-/*
- * 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.master.dto;
-
-import org.apache.dolphinscheduler.common.enums.Flag;
-import org.apache.dolphinscheduler.common.enums.Priority;
-import org.apache.dolphinscheduler.common.enums.TaskExecuteType;
-import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
-
-import java.util.Date;
-import java.util.Map;
-
-import lombok.Data;
-
-@Data
-public class TaskInstanceExecuteDto {
-
- private int id;
-
- private String name;
-
- private String taskType;
-
- private int processInstanceId;
-
- private long taskCode;
-
- private int taskDefinitionVersion;
-
- private String processInstanceName;
-
- private int taskGroupPriority;
-
- private TaskExecutionStatus state;
-
- private Date firstSubmitTime;
-
- private Date submitTime;
-
- private Date startTime;
-
- private Date endTime;
-
- private String host;
-
- private String executePath;
-
- private String logPath;
-
- private int retryTimes;
-
- private Flag alertFlag;
-
- private int pid;
-
- private String appLink;
-
- private Flag flag;
-
- private String duration;
-
- private int maxRetryTimes;
-
- private int retryInterval;
-
- private Priority taskInstancePriority;
-
- private Priority processInstancePriority;
-
- private String workerGroup;
-
- private Long environmentCode;
-
- private String environmentConfig;
-
- private int executorId;
-
- private String varPool;
-
- private String executorName;
-
- private Map<String, String> resources;
-
- private int delayTime;
-
- private String taskParams;
-
- private int dryRun;
-
- private int taskGroupId;
-
- private Integer cpuQuota;
-
- private Integer memoryMax;
-
- private TaskExecuteType taskExecuteType;
-}
diff --git
a/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/dto/WorkflowExecuteDto.java
b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/dto/WorkflowExecuteDto.java
deleted file mode 100644
index b4c7978dc8..0000000000
---
a/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/dto/WorkflowExecuteDto.java
+++ /dev/null
@@ -1,137 +0,0 @@
-/*
- * 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.master.dto;
-
-import org.apache.dolphinscheduler.common.enums.CommandType;
-import org.apache.dolphinscheduler.common.enums.FailureStrategy;
-import org.apache.dolphinscheduler.common.enums.Flag;
-import org.apache.dolphinscheduler.common.enums.Priority;
-import org.apache.dolphinscheduler.common.enums.TaskDependType;
-import org.apache.dolphinscheduler.common.enums.WarningType;
-import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
-
-import java.util.Collection;
-import java.util.Date;
-
-import lombok.Data;
-
-@Data
-public class WorkflowExecuteDto {
-
- private int id;
-
- private String name;
-
- private Long processDefinitionCode;
-
- private int processDefinitionVersion;
-
- private WorkflowExecutionStatus state;
-
- /**
- * recovery flag for failover
- */
- private Flag recovery;
-
- private Date startTime;
-
- private Date endTime;
-
- private int runTimes;
-
- private String host;
-
- private CommandType commandType;
-
- private String commandParam;
-
- /**
- * node depend type
- */
- private TaskDependType taskDependType;
-
- private int maxTryTimes;
-
- /**
- * failure strategy when task failed.
- */
- private FailureStrategy failureStrategy;
-
- private WarningType warningType;
-
- private Integer warningGroupId;
-
- private Date scheduleTime;
-
- private Date commandStartTime;
-
- /**
- * user define parameters string
- */
- private String globalParams;
-
- private int executorId;
-
- private String executorName;
-
- private String tenantCode;
-
- private String queue;
-
- /**
- * process is sub process
- */
- private Flag isSubProcess;
-
- /**
- * history command
- */
- private String historyCmd;
-
- /**
- * depend processes schedule time
- */
- private String dependenceScheduleTimes;
-
- private String duration;
-
- private Priority processInstancePriority;
-
- private String workerGroup;
-
- private Long environmentCode;
-
- private int timeout;
-
- private int tenantId;
-
- /**
- * varPool string
- */
- private String varPool;
-
- private int nextProcessInstanceId;
-
- private int dryRun;
-
- private Date restartTime;
-
- private boolean isBlocked;
-
- private Collection<TaskInstanceExecuteDto> taskInstances;
-}
diff --git a/dolphinscheduler-ui/src/service/modules/monitor/index.ts
b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/dto/WorkflowExecutorDTO.java
similarity index 58%
copy from dolphinscheduler-ui/src/service/modules/monitor/index.ts
copy to
dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/dto/WorkflowExecutorDTO.java
index a1ad1b3415..0235cc2376 100644
--- a/dolphinscheduler-ui/src/service/modules/monitor/index.ts
+++
b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/dto/WorkflowExecutorDTO.java
@@ -15,19 +15,35 @@
* limitations under the License.
*/
-import { axios } from '@/service/service'
-import type { ServerNodeType } from './types'
-
-export function queryDatabaseState(): any {
- return axios({
- url: '/monitor/databases',
- method: 'get'
- })
-}
+package org.apache.dolphinscheduler.extract.master.dto;
+
+import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
+
+import java.util.Date;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class WorkflowExecutorDTO {
+
+ private int id;
+
+ private String name;
+
+ private long projectCode;
+
+ private long workflowDefinitionCode;
+
+ private WorkflowExecutionStatus state;
+
+ private Date startTime;
+
+ private int runTimes;
-export function listMonitorServerNode(nodeType: ServerNodeType): any {
- return axios({
- url: `/monitor/${nodeType}`,
- method: 'get'
- })
}
diff --git a/dolphinscheduler-ui/src/service/modules/monitor/index.ts
b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowExecutorQueryRequest.java
similarity index 69%
copy from dolphinscheduler-ui/src/service/modules/monitor/index.ts
copy to
dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowExecutorQueryRequest.java
index a1ad1b3415..b0fe573781 100644
--- a/dolphinscheduler-ui/src/service/modules/monitor/index.ts
+++
b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowExecutorQueryRequest.java
@@ -15,19 +15,8 @@
* limitations under the License.
*/
-import { axios } from '@/service/service'
-import type { ServerNodeType } from './types'
+package org.apache.dolphinscheduler.extract.master.transportor;
-export function queryDatabaseState(): any {
- return axios({
- url: '/monitor/databases',
- method: 'get'
- })
-}
+public class WorkflowExecutorQueryRequest {
-export function listMonitorServerNode(nodeType: ServerNodeType): any {
- return axios({
- url: `/monitor/${nodeType}`,
- method: 'get'
- })
}
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/MonitorService.java
b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowExecutorQueryResponse.java
similarity index 51%
copy from
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/MonitorService.java
copy to
dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowExecutorQueryResponse.java
index 7c0e2d0bf9..d5a39f8d01 100644
---
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/MonitorService.java
+++
b/dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/WorkflowExecutorQueryResponse.java
@@ -15,30 +15,33 @@
* limitations under the License.
*/
-package org.apache.dolphinscheduler.api.service;
+package org.apache.dolphinscheduler.extract.master.transportor;
-import org.apache.dolphinscheduler.common.model.Server;
-import org.apache.dolphinscheduler.dao.entity.User;
-import org.apache.dolphinscheduler.dao.plugin.api.monitor.DatabaseMetrics;
-import org.apache.dolphinscheduler.registry.api.enums.RegistryNodeType;
+import org.apache.dolphinscheduler.extract.master.dto.WorkflowExecutorDTO;
import java.util.List;
-public interface MonitorService {
-
- /**
- * query database state
- *
- * @param loginUser login user
- * @return data base state
- */
- List<DatabaseMetrics> queryDatabaseState(User loginUser);
-
- /**
- * query server list
- *
- * @param nodeType RegistryNodeType
- * @return server information list
- */
- List<Server> listServer(RegistryNodeType nodeType);
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class WorkflowExecutorQueryResponse {
+
+ private boolean success;
+
+ private String message;
+
+ private List<WorkflowExecutorDTO> workflowExecutors;
+
+ public static WorkflowExecutorQueryResponse
success(List<WorkflowExecutorDTO> workflowExecutors) {
+ return new WorkflowExecutorQueryResponse(true, null,
workflowExecutors);
+ }
+
+ public static WorkflowExecutorQueryResponse fail(String message) {
+ return new WorkflowExecutorQueryResponse(false, message, null);
+ }
+
}
diff --git a/dolphinscheduler-ui/src/service/modules/monitor/index.ts
b/dolphinscheduler-extract/dolphinscheduler-extract-worker/src/main/java/org/apache/dolphinscheduler/extract/worker/ITaskExecutorQueryClient.java
similarity index 61%
copy from dolphinscheduler-ui/src/service/modules/monitor/index.ts
copy to
dolphinscheduler-extract/dolphinscheduler-extract-worker/src/main/java/org/apache/dolphinscheduler/extract/worker/ITaskExecutorQueryClient.java
index a1ad1b3415..2b833f9a26 100644
--- a/dolphinscheduler-ui/src/service/modules/monitor/index.ts
+++
b/dolphinscheduler-extract/dolphinscheduler-extract-worker/src/main/java/org/apache/dolphinscheduler/extract/worker/ITaskExecutorQueryClient.java
@@ -15,19 +15,17 @@
* limitations under the License.
*/
-import { axios } from '@/service/service'
-import type { ServerNodeType } from './types'
+package org.apache.dolphinscheduler.extract.worker;
-export function queryDatabaseState(): any {
- return axios({
- url: '/monitor/databases',
- method: 'get'
- })
-}
+import org.apache.dolphinscheduler.extract.base.RpcMethod;
+import org.apache.dolphinscheduler.extract.base.RpcService;
+import
org.apache.dolphinscheduler.extract.worker.transportor.TaskExecutorQueryRequest;
+import
org.apache.dolphinscheduler.extract.worker.transportor.TaskExecutorQueryResponse;
+
+@RpcService
+public interface ITaskExecutorQueryClient {
+
+ @RpcMethod
+ TaskExecutorQueryResponse queryTaskInstances(final
TaskExecutorQueryRequest request);
-export function listMonitorServerNode(nodeType: ServerNodeType): any {
- return axios({
- url: `/monitor/${nodeType}`,
- method: 'get'
- })
}
diff --git a/dolphinscheduler-ui/src/service/modules/monitor/index.ts
b/dolphinscheduler-extract/dolphinscheduler-extract-worker/src/main/java/org/apache/dolphinscheduler/extract/worker/transportor/TaskExecutorQueryRequest.java
similarity index 69%
copy from dolphinscheduler-ui/src/service/modules/monitor/index.ts
copy to
dolphinscheduler-extract/dolphinscheduler-extract-worker/src/main/java/org/apache/dolphinscheduler/extract/worker/transportor/TaskExecutorQueryRequest.java
index a1ad1b3415..25b2383041 100644
--- a/dolphinscheduler-ui/src/service/modules/monitor/index.ts
+++
b/dolphinscheduler-extract/dolphinscheduler-extract-worker/src/main/java/org/apache/dolphinscheduler/extract/worker/transportor/TaskExecutorQueryRequest.java
@@ -15,19 +15,8 @@
* limitations under the License.
*/
-import { axios } from '@/service/service'
-import type { ServerNodeType } from './types'
+package org.apache.dolphinscheduler.extract.worker.transportor;
-export function queryDatabaseState(): any {
- return axios({
- url: '/monitor/databases',
- method: 'get'
- })
-}
+public class TaskExecutorQueryRequest {
-export function listMonitorServerNode(nodeType: ServerNodeType): any {
- return axios({
- url: `/monitor/${nodeType}`,
- method: 'get'
- })
}
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/MonitorService.java
b/dolphinscheduler-extract/dolphinscheduler-extract-worker/src/main/java/org/apache/dolphinscheduler/extract/worker/transportor/TaskExecutorQueryResponse.java
similarity index 53%
copy from
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/MonitorService.java
copy to
dolphinscheduler-extract/dolphinscheduler-extract-worker/src/main/java/org/apache/dolphinscheduler/extract/worker/transportor/TaskExecutorQueryResponse.java
index 7c0e2d0bf9..17c5f0ce71 100644
---
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/MonitorService.java
+++
b/dolphinscheduler-extract/dolphinscheduler-extract-worker/src/main/java/org/apache/dolphinscheduler/extract/worker/transportor/TaskExecutorQueryResponse.java
@@ -15,30 +15,33 @@
* limitations under the License.
*/
-package org.apache.dolphinscheduler.api.service;
+package org.apache.dolphinscheduler.extract.worker.transportor;
-import org.apache.dolphinscheduler.common.model.Server;
-import org.apache.dolphinscheduler.dao.entity.User;
-import org.apache.dolphinscheduler.dao.plugin.api.monitor.DatabaseMetrics;
-import org.apache.dolphinscheduler.registry.api.enums.RegistryNodeType;
+import org.apache.dolphinscheduler.task.executor.dto.TaskExecutorDTO;
import java.util.List;
-public interface MonitorService {
-
- /**
- * query database state
- *
- * @param loginUser login user
- * @return data base state
- */
- List<DatabaseMetrics> queryDatabaseState(User loginUser);
-
- /**
- * query server list
- *
- * @param nodeType RegistryNodeType
- * @return server information list
- */
- List<Server> listServer(RegistryNodeType nodeType);
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class TaskExecutorQueryResponse {
+
+ private boolean success;
+
+ private String message;
+
+ private List<TaskExecutorDTO> taskExecutors;
+
+ public static TaskExecutorQueryResponse success(List<TaskExecutorDTO>
taskExecutors) {
+ return new TaskExecutorQueryResponse(true, null, taskExecutors);
+ }
+
+ public static TaskExecutorQueryResponse fail(String message) {
+ return new TaskExecutorQueryResponse(false, message, null);
+ }
+
}
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/WorkflowEngine.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/WorkflowEngine.java
index a0029b417a..e6e1bb79ba 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/WorkflowEngine.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/WorkflowEngine.java
@@ -17,10 +17,15 @@
package org.apache.dolphinscheduler.server.master.engine;
+import org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
+import org.apache.dolphinscheduler.extract.master.dto.WorkflowExecutorDTO;
import org.apache.dolphinscheduler.server.master.engine.command.CommandEngine;
import
org.apache.dolphinscheduler.server.master.engine.executor.LogicTaskEngineDelegator;
import
org.apache.dolphinscheduler.server.master.engine.task.dispatcher.WorkerGroupDispatcherCoordinator;
+import java.util.List;
+import java.util.stream.Collectors;
+
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
@@ -42,6 +47,9 @@ public class WorkflowEngine implements AutoCloseable {
@Autowired
private LogicTaskEngineDelegator logicTaskEngineDelegator;
+ @Autowired
+ private IWorkflowRepository workflowRepository;
+
public void start() {
workflowEventBusCoordinator.start();
@@ -55,6 +63,24 @@ public class WorkflowEngine implements AutoCloseable {
log.info("WorkflowEngine started");
}
+ public List<WorkflowExecutorDTO> queryWorkflowExecutors() {
+ return workflowRepository.getAll()
+ .stream()
+ .map(runnable -> {
+ WorkflowInstance wi = runnable.getWorkflowInstance();
+ return WorkflowExecutorDTO.builder()
+ .id(wi.getId())
+ .name(wi.getName())
+ .projectCode(wi.getProjectCode())
+
.workflowDefinitionCode(wi.getWorkflowDefinitionCode())
+ .state(wi.getState())
+ .startTime(wi.getStartTime())
+ .runTimes(wi.getRunTimes())
+ .build();
+ })
+ .collect(Collectors.toList());
+ }
+
@Override
public void close() throws Exception {
try (
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/LogicTaskEngineDelegator.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/LogicTaskEngineDelegator.java
index da76deec16..1ee95d2b1a 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/LogicTaskEngineDelegator.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/LogicTaskEngineDelegator.java
@@ -20,8 +20,11 @@ package
org.apache.dolphinscheduler.server.master.engine.executor;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
import org.apache.dolphinscheduler.task.executor.ITaskExecutor;
import org.apache.dolphinscheduler.task.executor.TaskEngine;
+import org.apache.dolphinscheduler.task.executor.dto.TaskExecutorDTO;
import
org.apache.dolphinscheduler.task.executor.eventbus.ITaskExecutorLifecycleEventReporter;
+import java.util.List;
+
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
@@ -63,6 +66,10 @@ public class LogicTaskEngineDelegator implements
AutoCloseable {
taskEngine.pauseTask(taskInstanceId);
}
+ public List<TaskExecutorDTO> queryTaskExecutors() {
+ return taskEngine.queryTaskExecutors();
+ }
+
public void ackLogicTaskExecutionEvent(final
ITaskExecutorLifecycleEventReporter.TaskExecutorLifecycleEventAck
taskExecutorLifecycleEventAck) {
logicTaskExecutorEventReporter.receiveTaskExecutorLifecycleEventACK(taskExecutorLifecycleEventAck);
}
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/runnable/TaskExecutionContextBuilder.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/runnable/TaskExecutionContextBuilder.java
index 9f331db11d..d77edcb69b 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/runnable/TaskExecutionContextBuilder.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/runnable/TaskExecutionContextBuilder.java
@@ -93,6 +93,8 @@ public class TaskExecutionContextBuilder {
*/
public TaskExecutionContextBuilder buildProcessInstanceRelatedInfo(final
WorkflowInstance workflowInstance) {
taskExecutionContext.setWorkflowInstanceId(workflowInstance.getId());
+
taskExecutionContext.setWorkflowInstanceName(workflowInstance.getName());
+ taskExecutionContext.setProjectCode(workflowInstance.getProjectCode());
taskExecutionContext.setScheduleTime(DateUtils.dateToTimeStamp(workflowInstance.getScheduleTime()));
taskExecutionContext.setGlobalParams(workflowInstance.getGlobalParams());
taskExecutionContext.setExecutorId(workflowInstance.getExecutorId());
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/LogicTaskExecutorQueryClient.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/LogicTaskExecutorQueryClient.java
new file mode 100644
index 0000000000..271e3fbee5
--- /dev/null
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/LogicTaskExecutorQueryClient.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dolphinscheduler.server.master.rpc;
+
+import org.apache.dolphinscheduler.extract.worker.ITaskExecutorQueryClient;
+import
org.apache.dolphinscheduler.extract.worker.transportor.TaskExecutorQueryRequest;
+import
org.apache.dolphinscheduler.extract.worker.transportor.TaskExecutorQueryResponse;
+import
org.apache.dolphinscheduler.server.master.engine.executor.LogicTaskEngineDelegator;
+import org.apache.dolphinscheduler.task.executor.dto.TaskExecutorDTO;
+
+import org.apache.commons.lang3.exception.ExceptionUtils;
+
+import java.util.List;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.springframework.stereotype.Service;
+
+@Slf4j
+@Service
+public class LogicTaskExecutorQueryClient implements ITaskExecutorQueryClient {
+
+ private final LogicTaskEngineDelegator logicTaskEngineDelegator;
+
+ public LogicTaskExecutorQueryClient(LogicTaskEngineDelegator
logicTaskEngineDelegator) {
+ this.logicTaskEngineDelegator = logicTaskEngineDelegator;
+ }
+
+ @Override
+ public TaskExecutorQueryResponse
queryTaskInstances(TaskExecutorQueryRequest request) {
+ try {
+ List<TaskExecutorDTO> tasks =
logicTaskEngineDelegator.queryTaskExecutors();
+ return TaskExecutorQueryResponse.success(tasks);
+ } catch (Exception ex) {
+ log.error("Query running task instances failed", ex);
+ return TaskExecutorQueryResponse.fail(
+ "Query running task instances failed: " +
ExceptionUtils.getMessage(ex));
+ }
+ }
+
+}
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/WorkflowExecutorQueryClient.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/WorkflowExecutorQueryClient.java
new file mode 100644
index 0000000000..e0a75d72ba
--- /dev/null
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/WorkflowExecutorQueryClient.java
@@ -0,0 +1,52 @@
+/*
+ * 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.server.master.rpc;
+
+import org.apache.dolphinscheduler.extract.master.IWorkflowExecutorQueryClient;
+import
org.apache.dolphinscheduler.extract.master.transportor.WorkflowExecutorQueryRequest;
+import
org.apache.dolphinscheduler.extract.master.transportor.WorkflowExecutorQueryResponse;
+import org.apache.dolphinscheduler.server.master.engine.WorkflowEngine;
+
+import org.apache.commons.lang3.exception.ExceptionUtils;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.springframework.stereotype.Service;
+
+@Slf4j
+@Service
+public class WorkflowExecutorQueryClient implements
IWorkflowExecutorQueryClient {
+
+ private final WorkflowEngine workflowEngine;
+
+ public WorkflowExecutorQueryClient(WorkflowEngine workflowEngine) {
+ this.workflowEngine = workflowEngine;
+ }
+
+ @Override
+ public WorkflowExecutorQueryResponse
queryWorkflowExecutors(WorkflowExecutorQueryRequest request) {
+ try {
+ return
WorkflowExecutorQueryResponse.success(workflowEngine.queryWorkflowExecutors());
+ } catch (Exception ex) {
+ log.error("Query running workflow instances failed", ex);
+ return WorkflowExecutorQueryResponse.fail(
+ "Query running workflow instances failed: " +
ExceptionUtils.getMessage(ex));
+ }
+ }
+
+}
diff --git
a/dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/ITaskEngine.java
b/dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/ITaskEngine.java
index f77a13ddac..d90abb0389 100644
---
a/dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/ITaskEngine.java
+++
b/dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/ITaskEngine.java
@@ -17,9 +17,12 @@
package org.apache.dolphinscheduler.task.executor;
+import org.apache.dolphinscheduler.task.executor.dto.TaskExecutorDTO;
import
org.apache.dolphinscheduler.task.executor.exceptions.TaskExecutorNotFoundException;
import
org.apache.dolphinscheduler.task.executor.exceptions.TaskExecutorRuntimeException;
+import java.util.List;
+
/**
* The TaskEngine interface used to responsible for task runtime.
*/
@@ -30,6 +33,13 @@ public interface ITaskEngine extends AutoCloseable {
*/
void start();
+ /**
+ * Query all task executors info.
+ *
+ * @return list of task executor DTOs
+ */
+ List<TaskExecutorDTO> queryTaskExecutors();
+
/**
* Submit the task to the task engine.
*
diff --git
a/dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/TaskEngine.java
b/dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/TaskEngine.java
index cf358fd9ea..9491eef638 100644
---
a/dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/TaskEngine.java
+++
b/dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/TaskEngine.java
@@ -17,8 +17,11 @@
package org.apache.dolphinscheduler.task.executor;
+import org.apache.dolphinscheduler.common.utils.DateUtils;
+import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
import
org.apache.dolphinscheduler.task.executor.container.ITaskExecutorContainer;
import
org.apache.dolphinscheduler.task.executor.container.ITaskExecutorContainerProvider;
+import org.apache.dolphinscheduler.task.executor.dto.TaskExecutorDTO;
import
org.apache.dolphinscheduler.task.executor.eventbus.ITaskExecutorEventBusCoordinator;
import
org.apache.dolphinscheduler.task.executor.events.TaskExecutorDispatchedLifecycleEvent;
import
org.apache.dolphinscheduler.task.executor.events.TaskExecutorKillLifecycleEvent;
@@ -27,6 +30,9 @@ import
org.apache.dolphinscheduler.task.executor.exceptions.TaskExecutorNotFound
import
org.apache.dolphinscheduler.task.executor.exceptions.TaskExecutorRuntimeException;
import org.apache.dolphinscheduler.task.executor.log.TaskExecutorMDCUtils;
+import java.util.List;
+import java.util.stream.Collectors;
+
import lombok.extern.slf4j.Slf4j;
@Slf4j
@@ -80,6 +86,24 @@ public class TaskEngine implements ITaskEngine {
}
}
+ @Override
+ public List<TaskExecutorDTO> queryTaskExecutors() {
+ return taskExecutorRepository.getAll().stream()
+ .map(executor -> {
+ TaskExecutionContext ctx =
executor.getTaskExecutionContext();
+ return TaskExecutorDTO.builder()
+ .id(ctx.getTaskInstanceId())
+ .name(ctx.getTaskName())
+ .taskType(ctx.getTaskType())
+ .projectCode(ctx.getProjectCode())
+ .workflowInstanceId(ctx.getWorkflowInstanceId())
+
.workflowInstanceName(ctx.getWorkflowInstanceName())
+
.startTime(DateUtils.timeStampToDate(ctx.getStartTime()))
+ .build();
+ })
+ .collect(Collectors.toList());
+ }
+
@Override
public void close() {
try (final ITaskExecutorEventBusCoordinator ignore1 =
taskExecutorEventBusCoordinator) {
diff --git a/dolphinscheduler-ui/src/service/modules/monitor/index.ts
b/dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/dto/TaskExecutorDTO.java
similarity index 62%
copy from dolphinscheduler-ui/src/service/modules/monitor/index.ts
copy to
dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/dto/TaskExecutorDTO.java
index a1ad1b3415..0d34faa330 100644
--- a/dolphinscheduler-ui/src/service/modules/monitor/index.ts
+++
b/dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/dto/TaskExecutorDTO.java
@@ -15,19 +15,33 @@
* limitations under the License.
*/
-import { axios } from '@/service/service'
-import type { ServerNodeType } from './types'
-
-export function queryDatabaseState(): any {
- return axios({
- url: '/monitor/databases',
- method: 'get'
- })
-}
+package org.apache.dolphinscheduler.task.executor.dto;
+
+import java.util.Date;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class TaskExecutorDTO {
+
+ private int id;
+
+ private String name;
+
+ private String taskType;
+
+ private long projectCode;
+
+ private int workflowInstanceId;
+
+ private String workflowInstanceName;
+
+ private Date startTime;
-export function listMonitorServerNode(nodeType: ServerNodeType): any {
- return axios({
- url: `/monitor/${nodeType}`,
- method: 'get'
- })
}
diff --git
a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java
b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java
index 94b4afd2ab..85d6f68219 100644
---
a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java
+++
b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java
@@ -75,6 +75,10 @@ public class TaskExecutionContext implements Serializable {
private int workflowInstanceId;
+ private String workflowInstanceName;
+
+ private Long projectCode;
+
private long scheduleTime;
private String globalParams;
@@ -124,8 +128,6 @@ public class TaskExecutionContext implements Serializable {
private int dispatchFailTimes;
- private boolean failover;
-
private final long firstDispatchTime = System.currentTimeMillis();
public int increaseDispatchFailTimes() {
diff --git a/dolphinscheduler-ui/src/locales/en_US/monitor.ts
b/dolphinscheduler-ui/src/locales/en_US/monitor.ts
index 5443153432..4eaeceeec5 100644
--- a/dolphinscheduler-ui/src/locales/en_US/monitor.ts
+++ b/dolphinscheduler-ui/src/locales/en_US/monitor.ts
@@ -25,6 +25,12 @@ export default {
directory_detail: 'Directory Detail',
host: 'Host',
directory: 'Directory',
+ running_workflows: 'Running Workflows',
+ workflow_name: 'Workflow Instance',
+ workflow_state: 'State',
+ workflow_start_time: 'Start Time',
+ workflow_run_times: 'Run Times',
+ running_tasks: 'Running Tasks',
master_no_data_result_title: 'No Master Nodes Exist',
master_no_data_result_desc:
'Currently, there are no master nodes exist, please create a master node
and refresh this page'
@@ -39,6 +45,11 @@ export default {
directory_detail: 'Directory Detail',
host: 'Host',
directory: 'Directory',
+ running_tasks: 'Running Tasks',
+ task_name: 'Task Name',
+ task_type: 'Task Type',
+ task_workflow_instance: 'Workflow Instance',
+ task_start_time: 'Start Time',
worker_no_data_result_title: 'No Worker Nodes Exist',
worker_no_data_result_desc:
'Currently, there are no worker nodes exist, please create a worker node
and refresh this page'
diff --git a/dolphinscheduler-ui/src/locales/zh_CN/monitor.ts
b/dolphinscheduler-ui/src/locales/zh_CN/monitor.ts
index ef78068a61..f305664a1d 100644
--- a/dolphinscheduler-ui/src/locales/zh_CN/monitor.ts
+++ b/dolphinscheduler-ui/src/locales/zh_CN/monitor.ts
@@ -25,6 +25,12 @@ export default {
directory_detail: '目录详情',
host: '主机',
directory: '注册目录',
+ running_workflows: '运行中的工作流',
+ workflow_name: '工作流实例',
+ workflow_state: '状态',
+ workflow_start_time: '开始时间',
+ workflow_run_times: '运行次数',
+ running_tasks: '运行中的任务',
master_no_data_result_title: 'Master节点不存在',
master_no_data_result_desc:
'目前没有任何Master节点，请先创建Master节点，再访问该页面'
@@ -39,6 +45,11 @@ export default {
directory_detail: '目录详情',
host: '主机',
directory: '注册目录',
+ running_tasks: '运行中的任务',
+ task_name: '任务名称',
+ task_type: '任务类型',
+ task_workflow_instance: '工作流实例',
+ task_start_time: '开始时间',
worker_no_data_result_title: 'Worker节点不存在',
worker_no_data_result_desc:
'目前没有任何Worker节点，请先创建Worker节点，再访问该页面'
diff --git a/dolphinscheduler-ui/src/service/modules/monitor/index.ts
b/dolphinscheduler-ui/src/service/modules/monitor/index.ts
index a1ad1b3415..ca3636c777 100644
--- a/dolphinscheduler-ui/src/service/modules/monitor/index.ts
+++ b/dolphinscheduler-ui/src/service/modules/monitor/index.ts
@@ -31,3 +31,19 @@ export function listMonitorServerNode(nodeType:
ServerNodeType): any {
method: 'get'
})
}
+
+export function queryWorkflowExecutors(masterAddress: string): any {
+ return axios({
+ url: '/monitor/masters/workflow-executors',
+ method: 'get',
+ params: { masterAddress }
+ })
+}
+
+export function queryTaskExecutors(serverAddress: string): any {
+ return axios({
+ url: '/monitor/workers/task-executors',
+ method: 'get',
+ params: { serverAddress }
+ })
+}
diff --git a/dolphinscheduler-ui/src/service/modules/monitor/types.ts
b/dolphinscheduler-ui/src/service/modules/monitor/types.ts
index 11a707789a..ef14a5e69d 100644
--- a/dolphinscheduler-ui/src/service/modules/monitor/types.ts
+++ b/dolphinscheduler-ui/src/service/modules/monitor/types.ts
@@ -49,11 +49,33 @@ interface WorkerNode extends ServerNode {
interface AlertNode extends MasterNode {}
+interface WorkflowExecutor {
+ id: number
+ name: string
+ projectCode: number
+ workflowDefinitionCode: number
+ state: string
+ startTime: string
+ runTimes: number
+}
+
+interface TaskExecutor {
+ id: number
+ name: string
+ taskType: string
+ projectCode: number
+ workflowInstanceId: number
+ workflowInstanceName: string
+ startTime: number
+}
+
export {
DatabaseRes,
MasterNode,
WorkerNode,
ServerNodeType,
ServerNode,
- AlertNode
+ AlertNode,
+ WorkflowExecutor,
+ TaskExecutor
}
diff --git a/dolphinscheduler-ui/src/views/monitor/servers/master/index.tsx
b/dolphinscheduler-ui/src/views/monitor/servers/master/index.tsx
index 0766681ddf..8412fe4c3e 100644
--- a/dolphinscheduler-ui/src/views/monitor/servers/master/index.tsx
+++ b/dolphinscheduler-ui/src/views/monitor/servers/master/index.tsx
@@ -24,16 +24,26 @@ import Card from '@/components/card'
import Result from '@/components/result'
import Gauge from '@/components/chart/modules/Gauge'
import MasterModal from './master-modal'
+import RunningWorkflowsModal from './running-workflows-modal'
+import RunningTasksModal from '../worker/running-tasks-modal'
+import { useUserStore } from '@/store/user/user'
import type { Ref } from 'vue'
import type { RowData } from 'naive-ui/es/data-table/src/interface'
import type { MasterNode } from '@/service/modules/monitor/types'
+import type { UserInfoRes } from '@/service/modules/users/types'
import { capitalize } from 'lodash'
const master = defineComponent({
name: 'master',
setup() {
const showModalRef = ref(false)
+ const showRunningRef = ref(false)
+ const showRunningTasksRef = ref(false)
+ const selectedMasterAddressRef = ref('')
const { t } = useI18n()
+ const userStore = useUserStore()
+ const IS_ADMIN =
+ (userStore.getUserInfo as UserInfoRes).userType === 'ADMIN_USER'
const { variables, getTableMaster } = useMaster()
const zkDirectoryRef: Ref<Array<RowData>> = ref([])
@@ -42,10 +52,28 @@ const master = defineComponent({
showModalRef.value = true
}
+ const clickRunningWorkflows = (item: MasterNode) => {
+ selectedMasterAddressRef.value = item.host + ':' + item.port
+ showRunningRef.value = true
+ }
+
+ const clickRunningTasks = (item: MasterNode) => {
+ selectedMasterAddressRef.value = item.host + ':' + item.port
+ showRunningTasksRef.value = true
+ }
+
const onConfirmModal = () => {
showModalRef.value = false
}
+ const onConfirmRunningModal = () => {
+ showRunningRef.value = false
+ }
+
+ const onConfirmRunningTasksModal = () => {
+ showRunningTasksRef.value = false
+ }
+
onMounted(() => {
getTableMaster()
})
@@ -54,14 +82,35 @@ const master = defineComponent({
t,
...toRefs(variables),
clickDetails,
+ clickRunningWorkflows,
+ clickRunningTasks,
onConfirmModal,
+ onConfirmRunningModal,
+ onConfirmRunningTasksModal,
showModalRef,
- zkDirectoryRef
+ showRunningRef,
+ showRunningTasksRef,
+ selectedMasterAddressRef,
+ zkDirectoryRef,
+ IS_ADMIN
}
},
render() {
- const { t, clickDetails, onConfirmModal, showModalRef, zkDirectoryRef } =
- this
+ const {
+ t,
+ clickDetails,
+ clickRunningWorkflows,
+ clickRunningTasks,
+ onConfirmModal,
+ onConfirmRunningModal,
+ onConfirmRunningTasksModal,
+ showModalRef,
+ showRunningRef,
+ showRunningTasksRef,
+ selectedMasterAddressRef,
+ zkDirectoryRef,
+ IS_ADMIN
+ } = this
const renderNodeServerStatusTag = (item: MasterNode) => {
const serverStatus = JSON.parse(item.heartBeatInfo)?.serverStatus
@@ -107,6 +156,22 @@ const master = defineComponent({
>
{t('monitor.master.directory_detail')}
</span>
+ {IS_ADMIN && (
+ <span
+ class={styles['link-btn']}
+ onClick={() => clickRunningWorkflows(item)}
+ >
+ {t('monitor.master.running_workflows')}
+ </span>
+ )}
+ {IS_ADMIN && (
+ <span
+ class={styles['link-btn']}
+ onClick={() => clickRunningTasks(item)}
+ >
+ {t('monitor.master.running_tasks')}
+ </span>
+ )}
</NSpace>
<NSpace>
<span>{`${t('monitor.master.create_time')}: ${
@@ -168,6 +233,16 @@ const master = defineComponent({
data={zkDirectoryRef}
onConfirmModal={onConfirmModal}
></MasterModal>
+ <RunningWorkflowsModal
+ showModal={showRunningRef}
+ masterAddress={selectedMasterAddressRef}
+ onConfirmModal={onConfirmRunningModal}
+ ></RunningWorkflowsModal>
+ <RunningTasksModal
+ showModal={showRunningTasksRef}
+ serverAddress={selectedMasterAddressRef}
+ onConfirmModal={onConfirmRunningTasksModal}
+ ></RunningTasksModal>
</>
)
}
diff --git
a/dolphinscheduler-ui/src/views/monitor/servers/master/running-workflows-modal.tsx
b/dolphinscheduler-ui/src/views/monitor/servers/master/running-workflows-modal.tsx
new file mode 100644
index 0000000000..297a89df5f
--- /dev/null
+++
b/dolphinscheduler-ui/src/views/monitor/servers/master/running-workflows-modal.tsx
@@ -0,0 +1,139 @@
+/*
+ * 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.
+ */
+
+import { defineComponent, ref, watch, h } from 'vue'
+import { useI18n } from 'vue-i18n'
+import { NDataTable, NSpin } from 'naive-ui'
+import { useRouter } from 'vue-router'
+import Modal from '@/components/modal'
+import { queryWorkflowExecutors } from '@/service/modules/monitor'
+import type { PropType } from 'vue'
+import type { WorkflowExecutor } from '@/service/modules/monitor/types'
+
+const props = {
+ showModal: {
+ type: Boolean as PropType<boolean>,
+ default: false
+ },
+ masterAddress: {
+ type: String as PropType<string>,
+ default: ''
+ }
+}
+
+const RunningWorkflowsModal = defineComponent({
+ props,
+ emits: ['confirmModal'],
+ setup(props, ctx) {
+ const { t } = useI18n()
+ const router = useRouter()
+ const loading = ref(false)
+ const data = ref<WorkflowExecutor[]>([])
+
+ const columns = [
+ {
+ title: '#',
+ key: 'index',
+ render: (_row: any, index: number) => index + 1,
+ width: 60
+ },
+ {
+ title: t('monitor.master.workflow_name'),
+ key: 'name',
+ render: (row: WorkflowExecutor) => {
+ return h(
+ 'a',
+ {
+ href:
`/projects/${row.projectCode}/workflow/instances/${row.id}`,
+ onClick: (e: MouseEvent) => {
+ e.preventDefault()
+ router.push({
+ path:
`/projects/${row.projectCode}/workflow/instances/${row.id}`
+ })
+ },
+ style: { color: '#2080f0', cursor: 'pointer' }
+ },
+ row.name
+ )
+ }
+ },
+ { title: t('monitor.master.workflow_state'), key: 'state' },
+ { title: t('monitor.master.workflow_start_time'), key: 'startTime' },
+ {
+ title: t('monitor.master.workflow_run_times'),
+ key: 'runTimes',
+ width: 100
+ }
+ ]
+
+ const fetchData = async () => {
+ if (!props.masterAddress) return
+ loading.value = true
+ try {
+ const res = await queryWorkflowExecutors(props.masterAddress)
+ data.value = res || []
+ } catch (e) {
+ data.value = []
+ } finally {
+ loading.value = false
+ }
+ }
+
+ watch(
+ () => props.showModal,
+ (val) => {
+ if (val) {
+ fetchData()
+ }
+ }
+ )
+
+ const onConfirm = () => {
+ ctx.emit('confirmModal')
+ }
+
+ return { t, columns, data, loading, onConfirm }
+ },
+ render() {
+ const { t, columns, data, loading, onConfirm } = this
+
+ return (
+ <Modal
+ title={t('monitor.master.running_workflows')}
+ show={this.showModal}
+ cancelShow={false}
+ onConfirm={onConfirm}
+ style={{ width: '800px' }}
+ >
+ {{
+ default: () => (
+ <NSpin show={loading}>
+ <NDataTable
+ columns={columns}
+ data={data}
+ striped
+ size={'small'}
+ />
+ </NSpin>
+ )
+ }}
+ </Modal>
+ )
+ }
+})
+
+export default RunningWorkflowsModal
diff --git a/dolphinscheduler-ui/src/views/monitor/servers/worker/index.tsx
b/dolphinscheduler-ui/src/views/monitor/servers/worker/index.tsx
index 34ff4786ac..44b1a540bb 100644
--- a/dolphinscheduler-ui/src/views/monitor/servers/worker/index.tsx
+++ b/dolphinscheduler-ui/src/views/monitor/servers/worker/index.tsx
@@ -24,16 +24,24 @@ import Card from '@/components/card'
import Result from '@/components/result'
import Gauge from '@/components/chart/modules/Gauge'
import WorkerModal from './worker-modal'
+import RunningTasksModal from './running-tasks-modal'
+import { useUserStore } from '@/store/user/user'
import type { Ref } from 'vue'
import type { RowData } from 'naive-ui/es/data-table/src/interface'
import type { WorkerNode } from '@/service/modules/monitor/types'
+import type { UserInfoRes } from '@/service/modules/users/types'
import { capitalize } from 'lodash'
const worker = defineComponent({
name: 'worker',
setup() {
const showModalRef = ref(false)
+ const showRunningRef = ref(false)
+ const selectedWorkerAddressRef = ref('')
const { t } = useI18n()
+ const userStore = useUserStore()
+ const IS_ADMIN =
+ (userStore.getUserInfo as UserInfoRes).userType === 'ADMIN_USER'
const { variables, getTableWorker } = useWorker()
const zkDirectoryRef: Ref<Array<RowData>> = ref([])
@@ -42,10 +50,19 @@ const worker = defineComponent({
showModalRef.value = true
}
+ const clickRunningTasks = (item: WorkerNode) => {
+ selectedWorkerAddressRef.value = item.host + ':' + item.port
+ showRunningRef.value = true
+ }
+
const onConfirmModal = () => {
showModalRef.value = false
}
+ const onConfirmRunningModal = () => {
+ showRunningRef.value = false
+ }
+
onMounted(() => {
getTableWorker()
})
@@ -54,14 +71,29 @@ const worker = defineComponent({
t,
...toRefs(variables),
clickDetails,
+ clickRunningTasks,
onConfirmModal,
+ onConfirmRunningModal,
showModalRef,
- zkDirectoryRef
+ showRunningRef,
+ selectedWorkerAddressRef,
+ zkDirectoryRef,
+ IS_ADMIN
}
},
render() {
- const { t, clickDetails, onConfirmModal, showModalRef, zkDirectoryRef } =
- this
+ const {
+ t,
+ clickDetails,
+ clickRunningTasks,
+ onConfirmModal,
+ onConfirmRunningModal,
+ showModalRef,
+ showRunningRef,
+ selectedWorkerAddressRef,
+ zkDirectoryRef,
+ IS_ADMIN
+ } = this
const renderNodeServerStatusTag = (item: WorkerNode) => {
const serverStatus = JSON.parse(item.heartBeatInfo)?.serverStatus
@@ -107,6 +139,14 @@ const worker = defineComponent({
>
{t('monitor.worker.directory_detail')}
</span>
+ {IS_ADMIN && (
+ <span
+ class={styles['link-btn']}
+ onClick={() => clickRunningTasks(item)}
+ >
+ {t('monitor.worker.running_tasks')}
+ </span>
+ )}
</NSpace>
<NSpace>
<span>{`${t('monitor.worker.create_time')}: ${
@@ -182,6 +222,11 @@ const worker = defineComponent({
data={zkDirectoryRef}
onConfirmModal={onConfirmModal}
/>
+ <RunningTasksModal
+ showModal={showRunningRef}
+ serverAddress={selectedWorkerAddressRef}
+ onConfirmModal={onConfirmRunningModal}
+ />
</>
)
}
diff --git
a/dolphinscheduler-ui/src/views/monitor/servers/worker/running-tasks-modal.tsx
b/dolphinscheduler-ui/src/views/monitor/servers/worker/running-tasks-modal.tsx
new file mode 100644
index 0000000000..b43cb6d92d
--- /dev/null
+++
b/dolphinscheduler-ui/src/views/monitor/servers/worker/running-tasks-modal.tsx
@@ -0,0 +1,135 @@
+/*
+ * 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.
+ */
+
+import { defineComponent, ref, watch, h } from 'vue'
+import { useI18n } from 'vue-i18n'
+import { NDataTable, NSpin } from 'naive-ui'
+import { useRouter } from 'vue-router'
+import Modal from '@/components/modal'
+import { queryTaskExecutors } from '@/service/modules/monitor'
+import type { PropType } from 'vue'
+import type { TaskExecutor } from '@/service/modules/monitor/types'
+
+const props = {
+ showModal: {
+ type: Boolean as PropType<boolean>,
+ default: false
+ },
+ serverAddress: {
+ type: String as PropType<string>,
+ default: ''
+ }
+}
+
+const RunningTasksModal = defineComponent({
+ props,
+ emits: ['confirmModal'],
+ setup(props, ctx) {
+ const { t } = useI18n()
+ const router = useRouter()
+ const loading = ref(false)
+ const data = ref<TaskExecutor[]>([])
+
+ const columns = [
+ {
+ title: '#',
+ key: 'index',
+ render: (_row: any, index: number) => index + 1,
+ width: 60
+ },
+ { title: t('monitor.worker.task_name'), key: 'name' },
+ { title: t('monitor.worker.task_type'), key: 'taskType' },
+ {
+ title: t('monitor.worker.task_workflow_instance'),
+ key: 'workflowInstanceName',
+ render: (row: TaskExecutor) => {
+ return h(
+ 'a',
+ {
+ href:
`/projects/${row.projectCode}/workflow/instances/${row.workflowInstanceId}`,
+ onClick: (e: MouseEvent) => {
+ e.preventDefault()
+ router.push({
+ path:
`/projects/${row.projectCode}/workflow/instances/${row.workflowInstanceId}`
+ })
+ },
+ style: { color: '#2080f0', cursor: 'pointer' }
+ },
+ row.workflowInstanceName || row.workflowInstanceId
+ )
+ }
+ },
+ { title: t('monitor.worker.task_start_time'), key: 'startTime' }
+ ]
+
+ const fetchData = async () => {
+ if (!props.serverAddress) return
+ loading.value = true
+ try {
+ const res = await queryTaskExecutors(props.serverAddress)
+ data.value = res || []
+ } catch (e) {
+ data.value = []
+ } finally {
+ loading.value = false
+ }
+ }
+
+ watch(
+ () => props.showModal,
+ (val) => {
+ if (val) {
+ fetchData()
+ }
+ }
+ )
+
+ const onConfirm = () => {
+ ctx.emit('confirmModal')
+ }
+
+ return { t, columns, data, loading, onConfirm }
+ },
+ render() {
+ const { t, columns, data, loading, onConfirm } = this
+
+ return (
+ <Modal
+ title={t('monitor.worker.running_tasks')}
+ show={this.showModal}
+ cancelShow={false}
+ onConfirm={onConfirm}
+ style={{ width: '800px' }}
+ >
+ {{
+ default: () => (
+ <NSpin show={loading}>
+ <NDataTable
+ columns={columns}
+ data={data}
+ striped
+ size={'small'}
+ />
+ </NSpin>
+ )
+ }}
+ </Modal>
+ )
+ }
+})
+
+export default RunningTasksModal
diff --git
a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/executor/PhysicalTaskEngineDelegator.java
b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/executor/PhysicalTaskEngineDelegator.java
index 67aba113e6..c7a692e127 100644
---
a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/executor/PhysicalTaskEngineDelegator.java
+++
b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/executor/PhysicalTaskEngineDelegator.java
@@ -20,9 +20,11 @@ package org.apache.dolphinscheduler.server.worker.executor;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
import org.apache.dolphinscheduler.task.executor.ITaskExecutor;
import org.apache.dolphinscheduler.task.executor.TaskEngine;
+import org.apache.dolphinscheduler.task.executor.dto.TaskExecutorDTO;
import
org.apache.dolphinscheduler.task.executor.eventbus.ITaskExecutorLifecycleEventReporter;
import
org.apache.dolphinscheduler.task.executor.operations.TaskExecutorReassignMasterRequest;
+import java.util.List;
import java.util.Optional;
import lombok.extern.slf4j.Slf4j;
@@ -70,6 +72,10 @@ public class PhysicalTaskEngineDelegator implements
AutoCloseable {
taskEngine.pauseTask(taskInstanceId);
}
+ public List<TaskExecutorDTO> queryTaskExecutors() {
+ return taskEngine.queryTaskExecutors();
+ }
+
public void ackPhysicalTaskExecutorLifecycleEventACK(final
ITaskExecutorLifecycleEventReporter.TaskExecutorLifecycleEventAck
taskExecutorLifecycleEventAck) {
physicalTaskExecutorEventReporter.receiveTaskExecutorLifecycleEventACK(taskExecutorLifecycleEventAck);
}
diff --git
a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/rpc/PhysicalTaskExecutorQueryClient.java
b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/rpc/PhysicalTaskExecutorQueryClient.java
new file mode 100644
index 0000000000..0b97c97aeb
--- /dev/null
+++
b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/rpc/PhysicalTaskExecutorQueryClient.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dolphinscheduler.server.worker.rpc;
+
+import org.apache.dolphinscheduler.extract.worker.ITaskExecutorQueryClient;
+import
org.apache.dolphinscheduler.extract.worker.transportor.TaskExecutorQueryRequest;
+import
org.apache.dolphinscheduler.extract.worker.transportor.TaskExecutorQueryResponse;
+import
org.apache.dolphinscheduler.server.worker.executor.PhysicalTaskEngineDelegator;
+import org.apache.dolphinscheduler.task.executor.dto.TaskExecutorDTO;
+
+import org.apache.commons.lang3.exception.ExceptionUtils;
+
+import java.util.List;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.springframework.stereotype.Service;
+
+@Slf4j
+@Service
+public class PhysicalTaskExecutorQueryClient implements
ITaskExecutorQueryClient {
+
+ private final PhysicalTaskEngineDelegator physicalTaskEngineDelegator;
+
+ public PhysicalTaskExecutorQueryClient(PhysicalTaskEngineDelegator
physicalTaskEngineDelegator) {
+ this.physicalTaskEngineDelegator = physicalTaskEngineDelegator;
+ }
+
+ @Override
+ public TaskExecutorQueryResponse
queryTaskInstances(TaskExecutorQueryRequest request) {
+ try {
+ List<TaskExecutorDTO> tasks =
physicalTaskEngineDelegator.queryTaskExecutors();
+ return TaskExecutorQueryResponse.success(tasks);
+ } catch (Exception ex) {
+ log.error("Query running task instances failed", ex);
+ return TaskExecutorQueryResponse.fail(
+ "Query running task instances failed: " +
ExceptionUtils.getMessage(ex));
+ }
+ }
+
+}
