[Feature-18136][API] Support view the running task/workflow of a active master/worker at UI monitor page (#18138)

Author: ruanwenjun

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.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.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 Result<List<DatabaseMetrics>> queryDatabaseState(@Parameter(hidden = true
         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);
+    }
+
 }

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/MonitorService.java

@@ -20,7 +20,9 @@
 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);
 }

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 List<DatabaseMetrics> queryDatabaseState(User loginUser) {
     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();
+    }
+
 }

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.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()))

dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/IWorkflowExecutorQueryClient.java

@@ -0,0 +1,31 @@
+/*
+ * 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;
+
+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);
+
+}