feature: add flink webui proxy

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

@@ -0,0 +1,50 @@
+/*
+ * 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.core.service.ApplicationService;
+
+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.PathVariable;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestMethod;
+import org.springframework.web.bind.annotation.RestController;
+
+import javax.servlet.http.HttpServletRequest;
+
+import java.io.IOException;
+
+@Slf4j
+@Validated
+@RestController
+@RequestMapping("proxy")
+public class ProxyController {
+
+  @Autowired private ApplicationService applicationService;
+
+  @RequestMapping(
+      value = "flink-ui/{id}/**",
+      method = {RequestMethod.GET, RequestMethod.POST})
+  public ResponseEntity<?> proxyRequest(HttpServletRequest request, @PathVariable("id") Long id)
+      throws IOException {
+    return applicationService.proxyFlinkUI(request, id);
+  }
+}

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

@@ -599,6 +599,10 @@ private boolean needFillYarnQueueLabel(ExecutionMode mode) {
     return ExecutionMode.YARN_PER_JOB.equals(mode) || ExecutionMode.YARN_APPLICATION.equals(mode);
   }
 
+  public String getFlinkRestUrl() {
+    return "/proxy/flink-ui/" + id + "/";
+  }
+
   @Override
   public boolean equals(Object o) {
     if (this == o) {

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

@@ -28,8 +28,11 @@
 
 import com.baomidou.mybatisplus.core.metadata.IPage;
 import com.baomidou.mybatisplus.extension.service.IService;
+import org.springframework.http.ResponseEntity;
 import org.springframework.web.multipart.MultipartFile;
 
+import javax.servlet.http.HttpServletRequest;
+
 import java.io.IOException;
 import java.io.Serializable;
 import java.util.Collection;
@@ -131,4 +134,6 @@ List<Application> getByTeamIdAndExecutionModes(
   List<ApplicationReport> getYARNApplication(String appName);
 
   RestResponse buildApplication(Long appId, boolean forceBuild) throws Exception;
+
+  ResponseEntity<?> proxyFlinkUI(HttpServletRequest request, Long appId) throws IOException;
 }

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

@@ -83,6 +83,9 @@
 import org.apache.streampark.console.core.task.CheckpointProcessor;
 import org.apache.streampark.console.core.task.FlinkAppHttpWatcher;
 import org.apache.streampark.console.core.task.FlinkK8sWatcherWrapper;
+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.flink.client.FlinkClient;
 import org.apache.streampark.flink.client.bean.CancelRequest;
 import org.apache.streampark.flink.client.bean.CancelResponse;
@@ -99,6 +102,7 @@
 
 import org.apache.commons.io.FileUtils;
 import org.apache.commons.io.FilenameUtils;
+import org.apache.commons.io.IOUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.configuration.CoreOptions;
@@ -123,17 +127,29 @@
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.web.client.RestTemplateBuilder;
+import org.springframework.http.HttpEntity;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.http.client.ClientHttpResponse;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.annotation.Transactional;
+import org.springframework.web.client.DefaultResponseErrorHandler;
+import org.springframework.web.client.RestTemplate;
 import org.springframework.web.multipart.MultipartFile;
 
 import javax.annotation.Nonnull;
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
+import javax.servlet.http.HttpServletRequest;
 
+import java.io.ByteArrayOutputStream;
 import java.io.File;
 import java.io.IOException;
+import java.io.InputStream;
 import java.io.Serializable;
 import java.net.URI;
 import java.nio.charset.StandardCharsets;
@@ -144,6 +160,7 @@
 import java.util.Comparator;
 import java.util.Date;
 import java.util.EnumSet;
+import java.util.Enumeration;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -218,6 +235,10 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
 
   @Autowired private CheckpointProcessor checkpointProcessor;
 
+  @Autowired private MemberService memberService;
+
+  private final RestTemplate proxyRestTemplate;
+
   private static final int CPU_NUM = Math.max(2, Runtime.getRuntime().availableProcessors() * 4);
 
   private final ExecutorService bootstrapExecutor =
@@ -229,6 +250,20 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
           new LinkedBlockingQueue<>(),
           ThreadUtils.threadFactory("streampark-flink-app-bootstrap"));
 
+  @Autowired
+  public ApplicationServiceImpl(RestTemplateBuilder restTemplateBuilder) {
+    this.proxyRestTemplate =
+        restTemplateBuilder
+            .errorHandler(
+                new DefaultResponseErrorHandler() {
+                  @Override
+                  public void handleError(@Nonnull ClientHttpResponse response) {
+                    // Ignore errors in the Flink Web UI itself, such as 404 errors.
+                  }
+                })
+            .build();
+  }
+
   @PostConstruct
   public void resetOptionState() {
     this.baseMapper.resetOptionState();
@@ -2036,4 +2071,50 @@ private Tuple2<String, String> getNamespaceClusterId(Application application) {
     }
     return Tuple2.apply(k8sNamespace, clusterId);
   }
+
+  public ResponseEntity<?> proxyFlinkUI(HttpServletRequest request, Long appId) throws IOException {
+    ApiAlertException.throwIfTrue(appId == null, "Invalid operation, appId is null");
+
+    User currentUser = serviceHelper.getLoginUser();
+    Application app = getById(appId);
+    ApiAlertException.throwIfTrue(app == null, "Invalid operation, application is null");
+    if (!currentUser.getUserId().equals(app.getUserId())) {
+      Member member = memberService.findByUserName(app.getTeamId(), currentUser.getUsername());
+      ApiAlertException.throwIfTrue(
+          member == null,
+          "Permission denied, this job not created by the current user, And the job cannot be found in the current user's team.");
+    }
+
+    String originalUrl =
+        request.getRequestURI()
+            + (request.getQueryString() != null ? "?" + request.getQueryString() : "");
+
+    String jobManagerUrl = app.getJobManagerUrl();
+    if (jobManagerUrl == null) {
+      ResponseEntity.BodyBuilder builder = ResponseEntity.status(HttpStatus.SERVICE_UNAVAILABLE);
+      builder.body("The flink job manager url is not ready");
+      return builder.build();
+    }
+
+    String newUrl = jobManagerUrl + originalUrl.replace("/proxy/flink-ui/" + appId, "");
+
+    HttpHeaders headers = new HttpHeaders();
+    Enumeration<String> headerNames = request.getHeaderNames();
+    while (headerNames.hasMoreElements()) {
+      String headerName = headerNames.nextElement();
+      headers.set(headerName, request.getHeader(headerName));
+    }
+
+    byte[] body = null;
+    if (request.getInputStream().available() > 0) {
+      InputStream inputStream = request.getInputStream();
+      ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+      IOUtils.copy(inputStream, byteArrayOutputStream);
+      body = byteArrayOutputStream.toByteArray();
+    }
+
+    HttpEntity<?> requestEntity = new HttpEntity<>(body, headers);
+    return proxyRestTemplate.exchange(
+        newUrl, HttpMethod.valueOf(request.getMethod()), requestEntity, byte[].class);
+  }
 }

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/authentication/ShiroConfig.java

@@ -74,6 +74,7 @@ public ShiroFilterFactoryBean shiroFilterFactoryBean(SecurityManager securityMan
     filterChainDefinitionMap.put("/*.less", "anon");
     filterChainDefinitionMap.put("/*.ico", "anon");
     filterChainDefinitionMap.put("/", "anon");
+    filterChainDefinitionMap.put("/proxy/flink-ui/**", "anon");
     filterChainDefinitionMap.put("/**", "jwt");
 
     shiroFilterFactoryBean.setFilterChainDefinitionMap(filterChainDefinitionMap);