[Improve] dev: flink webUI proxy support.

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ApplicationController.java

@@ -214,17 +214,17 @@ public RestResponse backups(ApplicationBackUp backUp, RestRequest request) {
     }
 
     @PostMapping("opt_log")
-    @Permission(app = "#log.appId", team = "#log.teamId")
+    @Permission(app = "#applicationLog.appId", team = "#applicationLog.teamId")
     public RestResponse log(ApplicationLog applicationLog, RestRequest request) {
         IPage<ApplicationLog> applicationList = applicationLogService.getPage(applicationLog, request);
         return RestResponse.success(applicationList);
     }
 
-    @Permission(app = "#log.appId", team = "#log.teamId")
+    @Permission(app = "#applicationLog.appId", team = "#applicationLog.teamId")
     @PostMapping("delete/opt_log")
     @RequiresPermissions("app:delete")
-    public RestResponse deleteLog(Long id) {
-        Boolean deleted = applicationLogService.removeById(id);
+    public RestResponse deleteLog(ApplicationLog applicationLog) {
+        Boolean deleted = applicationLogService.delete(applicationLog);
         return RestResponse.success(deleted);
     }
 

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ProxyController.java

@@ -0,0 +1,122 @@
+/*
+ * 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.streampark.console.core.controller;
+
+import org.apache.streampark.console.base.exception.ApiAlertException;
+import org.apache.streampark.console.core.entity.Application;
+import org.apache.streampark.console.core.entity.ApplicationLog;
+import org.apache.streampark.console.core.enums.UserTypeEnum;
+import org.apache.streampark.console.core.service.ApplicationLogService;
+import org.apache.streampark.console.core.service.ProxyService;
+import org.apache.streampark.console.core.service.application.ApplicationManageService;
+import org.apache.streampark.console.core.util.ServiceHelper;
+import org.apache.streampark.console.system.entity.Member;
+import org.apache.streampark.console.system.entity.User;
+import org.apache.streampark.console.system.service.MemberService;
+import org.apache.streampark.console.system.service.UserService;
+
+import org.apache.shiro.authz.annotation.RequiresPermissions;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.ResponseEntity;
+import org.springframework.validation.annotation.Validated;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import javax.servlet.http.HttpServletRequest;
+
+@Slf4j
+@Validated
+@RestController
+@RequestMapping("proxy")
+public class ProxyController {
+
+    @Autowired
+    private ProxyService proxyService;
+
+    @Autowired
+    private ApplicationManageService applicationManageService;
+
+    @Autowired
+    private ApplicationLogService logService;
+
+    @Autowired
+    private MemberService memberService;
+
+    @Autowired
+    private UserService userService;
+
+    @GetMapping("{type}/{id}/assets/**")
+    public ResponseEntity<?> proxyFlinkAssets(HttpServletRequest request, @PathVariable("type") String type,
+                                              @PathVariable("id") Long id) throws Exception {
+        return proxy(type, request, id);
+    }
+
+    @GetMapping("{type}/{id}/**")
+    @RequiresPermissions("app:view")
+    public ResponseEntity<?> proxyFlink(HttpServletRequest request, @PathVariable("type") String type,
+                                        @PathVariable("id") Long id) throws Exception {
+        return proxy(type, request, id);
+    }
+
+    private ResponseEntity<?> proxy(String type, HttpServletRequest request, Long id) throws Exception {
+        ApplicationLog log;
+        Application app;
+
+        switch (type) {
+            case "flink":
+                app = applicationManageService.getApp(id);
+                checkProxyApp(app);
+                return proxyService.proxyFlink(request, app);
+            case "cluster":
+                return proxyService.proxyCluster(request, id);
+            case "history":
+                log = logService.getById(id);
+                checkProxyAppLog(log);
+                return proxyService.proxyHistory(request, log);
+            case "yarn":
+                log = logService.getById(id);
+                checkProxyAppLog(log);
+                return proxyService.proxyYarn(request, log);
+            default:
+                return ResponseEntity.notFound().build();
+        }
+    }
+
+    private void checkProxyApp(Application app) {
+        ApiAlertException.throwIfNull(app, "Invalid operation, application is invalid.");
+
+        User user = ServiceHelper.getLoginUser();
+        ApiAlertException.throwIfNull(user, "Permission denied, please login first.");
+
+        if (user.getUserType() != UserTypeEnum.ADMIN) {
+            Member member = memberService.getByTeamIdUserName(app.getTeamId(), user.getUsername());
+            ApiAlertException.throwIfNull(member,
+                "Permission denied, this job not created by the current user, And the job cannot be found in the current user's team.");
+        }
+    }
+
+    private void checkProxyAppLog(ApplicationLog log) {
+        ApiAlertException.throwIfNull(log, "Invalid operation, The application log not found.");
+        Application app = applicationManageService.getById(log.getAppId());
+        checkProxyApp(app);
+    }
+}

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/ApplicationBackUp.java

@@ -47,6 +47,8 @@ public class ApplicationBackUp {
 
     private transient boolean backup;
 
+    private transient String teamId;
+
     public ApplicationBackUp() {
     }
 

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/ApplicationLog.java

@@ -48,4 +48,6 @@ public class ApplicationLog {
     private String exception;
     /** The user who operates the application */
     private Long userId;
+
+    private transient String teamId;
 }

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/FlinkSql.java

@@ -61,6 +61,8 @@ public class FlinkSql {
     /** dependency diff */
     private transient boolean dependencyDifference = false;
 
+    private transient Long teamId;
+
     public FlinkSql() {
     }
 

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/Savepoint.java

@@ -50,4 +50,6 @@ public class Savepoint {
     private Date triggerTime;
 
     private Date createTime;
+
+    private transient Long teamId;
 }

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/ApplicationLogService.java

@@ -41,4 +41,6 @@ public interface ApplicationLogService extends IService<ApplicationLog> {
      * @param appId The id of the application to be removed
      */
     void removeByAppId(Long appId);
+
+    Boolean delete(ApplicationLog applicationLog);
 }

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/ProxyService.java

@@ -0,0 +1,36 @@
+/*
+ * 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.streampark.console.core.service;
+
+import org.apache.streampark.console.core.entity.Application;
+import org.apache.streampark.console.core.entity.ApplicationLog;
+
+import org.springframework.http.ResponseEntity;
+
+import javax.servlet.http.HttpServletRequest;
+
+public interface ProxyService {
+
+    ResponseEntity<?> proxyFlink(HttpServletRequest request, Application app) throws Exception;
+
+    ResponseEntity<?> proxyYarn(HttpServletRequest request, ApplicationLog log) throws Exception;
+
+    ResponseEntity<?> proxyHistory(HttpServletRequest request, ApplicationLog log) throws Exception;
+
+    ResponseEntity<?> proxyCluster(HttpServletRequest request, Long clusterId) throws Exception;
+}

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationLogServiceImpl.java

@@ -54,4 +54,9 @@ public void removeByAppId(Long appId) {
             .eq(ApplicationLog::getAppId, appId);
         this.remove(queryWrapper);
     }
+
+    @Override
+    public Boolean delete(ApplicationLog applicationLog) {
+        return removeById(applicationLog.getId());
+    }
 }