add ingress configuration

streamx-console/streamx-console-service/src/assembly/script/final.sql

@@ -106,6 +106,8 @@ CREATE TABLE `t_flink_app` (
 `K8S_TM_POD_TEMPLATE` text COLLATE utf8mb4_general_ci,
 `K8S_HADOOP_INTEGRATION` tinyint(1) DEFAULT '0',
 `FLINK_CLUSTER_ID` bigint DEFAULT NULL,
+`INGRESS_TEMPLATE` text COLLATE utf8mb4_general_ci,
+`DEFAULT_MODE_INGRESS` text COLLATE utf8mb4_general_ci,
 PRIMARY KEY (`ID`) USING BTREE,
 KEY `INX_STATE` (`STATE`) USING BTREE,
 KEY `INX_JOB_TYPE` (`JOB_TYPE`) USING BTREE,
@@ -116,7 +118,7 @@ KEY `INX_TRACK` (`TRACKING`) USING BTREE
 -- Records of t_flink_app
 -- ----------------------------
 BEGIN;
-INSERT INTO `t_flink_app` VALUES (100000, 2, 4, NULL, NULL, 'Flink SQL Demo', NULL, NULL, NULL, NULL, NULL, NULL , NULL, 100000, NULL, 1, NULL, NULL, NULL, NULL, NULL, NULL, '0', 0, NULL, NULL, NULL, NULL, NULL, NULL, 'Flink SQL Demo', 0, NULL, 0, NULL, NULL, NULL, NULL, NULL, NULL, 0, 0, NOW(), NULL, 1, 1, NULL, NULL, NULL, NULL, NULL, NULL, 0, NULL);
+INSERT INTO `t_flink_app` VALUES (100000, 2, 4, NULL, NULL, 'Flink SQL Demo', NULL, NULL, NULL, NULL, NULL, NULL , NULL, 100000, NULL, 1, NULL, NULL, NULL, NULL, NULL, NULL, '0', 0, NULL, NULL, NULL, NULL, NULL, NULL, 'Flink SQL Demo', 0, NULL, 0, NULL, NULL, NULL, NULL, NULL, NULL, 0, 0, NOW(), NULL, 1, 1, NULL, NULL, NULL, NULL, NULL, NULL, 0, NULL, NULL, NULL);
 COMMIT;
 
 -- ----------------------------

streamx-console/streamx-console-service/src/assembly/script/upgrade.sql

@@ -310,4 +310,6 @@ alter table t_flink_app drop column ALERT_EMAIL;
 ALTER TABLE `t_flink_app` ADD COLUMN `OPTION_TIME` datetime DEFAULT NULL AFTER `CREATE_TIME`;
 ALTER TABLE t_setting modify column `VALUE` text ;
 INSERT INTO `t_setting` VALUES (14, 'docker.register.namespace', NULL, 'Docker Register Image namespace', 'Docker命名空间', 1);
+ALTER TABLE `t_flink_app` ADD COLUMN `INGRESS_TEMPLATE` text COLLATE utf8mb4_general_ci COMMENT 'ingress模版文件';
+ALTER TABLE `t_flink_app` ADD COLUMN `DEFAULT_MODE_INGRESS` text COLLATE utf8mb4_general_ci COMMENT '配置ingress的域名';
 -- ------------------------------------- version: 1.2.4 END ---------------------------------------

streamx-console/streamx-console-service/src/main/java/com/streamxhub/streamx/console/core/entity/Application.java

@@ -243,6 +243,9 @@ public class Application implements Serializable {
     private String k8sJmPodTemplate;
     private String k8sTmPodTemplate;
 
+    private String ingressTemplate;
+    private String defaultModeIngress;
+
     /**
      * 1: cicd (build from csv)
      * 2: upload (upload local jar job)
@@ -297,6 +300,22 @@ public class Application implements Serializable {
 
     private transient AppControl appControl;
 
+    public String getIngressTemplate() {
+        return ingressTemplate;
+    }
+
+    public void setIngressTemplate(String ingressTemplate) {
+        this.ingressTemplate = ingressTemplate;
+    }
+
+    public String getDefaultModeIngress() {
+        return defaultModeIngress;
+    }
+
+    public void setDefaultModeIngress(String defaultModeIngress) {
+        this.defaultModeIngress = defaultModeIngress;
+    }
+
     public void setK8sNamespace(String k8sNamespace) {
         this.k8sNamespace = StringUtils.isBlank(k8sNamespace) ? K8sFlinkConfig.DEFAULT_KUBERNETES_NAMESPACE() : k8sNamespace;
     }

streamx-console/streamx-console-service/src/main/java/com/streamxhub/streamx/console/core/service/impl/ApplBuildPipeServiceImpl.java

@@ -388,7 +388,8 @@ private BuildPipeline createPipelineInstance(@Nonnull Application app) {
                                 settingService.getDockerRegisterNamespace(),
                                 settingService.getDockerRegisterUser(),
                                 settingService.getDockerRegisterPassword()
-                        )
+                        ),
+                    app.getIngressTemplate()
                 );
                 log.info("Submit params to building pipeline : {}", k8sApplicationBuildRequest);
                 return FlinkK8sApplicationBuildPipeline.of(k8sApplicationBuildRequest);

streamx-console/streamx-console-service/src/main/java/com/streamxhub/streamx/console/core/service/impl/ApplicationServiceImpl.java

@@ -80,7 +80,9 @@
 import com.streamxhub.streamx.flink.kubernetes.K8sFlinkTrkMonitor;
 import com.streamxhub.streamx.flink.kubernetes.model.FlinkMetricCV;
 import com.streamxhub.streamx.flink.kubernetes.model.TrkId;
+import com.streamxhub.streamx.flink.kubernetes.network.FlinkJobIngress;
 import com.streamxhub.streamx.flink.packer.pipeline.BuildResult;
+import com.streamxhub.streamx.flink.packer.pipeline.DockerImageBuildResponse;
 import com.streamxhub.streamx.flink.packer.pipeline.ShadedBuildResponse;
 import com.streamxhub.streamx.flink.submit.FlinkSubmitter;
 import com.streamxhub.streamx.flink.submit.bean.CancelRequest;
@@ -1210,6 +1212,18 @@ public void start(Application appParam, boolean auto) throws Exception {
             extraParameter
         );
 
+        DockerImageBuildResponse result = buildResult.as(DockerImageBuildResponse.class);
+
+        String ingressTemplates = application.getIngressTemplate();
+        String domainName = application.getDefaultModeIngress();
+        if (StringUtils.isNotBlank(ingressTemplates)) {
+            String ingressOutput = result.workspacePath() + "/ingress.yaml";
+            FlinkJobIngress.configureIngress(ingressOutput);
+        }
+        if (StringUtils.isNotBlank(domainName)) {
+            FlinkJobIngress.configureIngress(domainName, application.getClusterId(), application.getK8sNamespace());
+        }
+
         CompletableFuture<SubmitResponse> future = CompletableFuture.supplyAsync(() -> FlinkSubmitter.submit(submitRequest), executorService);
 
         startFutureMap.put(application.getId(), future);

streamx-console/streamx-console-service/src/main/java/com/streamxhub/streamx/console/core/task/K8sFlinkChangeEventListener.java

@@ -37,6 +37,7 @@
 import com.streamxhub.streamx.flink.kubernetes.model.FlinkMetricCV;
 import com.streamxhub.streamx.flink.kubernetes.model.JobStatusCV;
 import com.streamxhub.streamx.flink.kubernetes.model.TrkId;
+import com.streamxhub.streamx.flink.kubernetes.network.FlinkJobIngress;
 import com.streamxhub.streamx.flink.kubernetes.watcher.FlinkJobStatusWatcher;
 
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
@@ -111,6 +112,7 @@ public void persistentK8sFlinkJobStatusChange(FlinkJobStatusChangeEvent event) {
         FlinkAppState state = FlinkAppState.of(app.getState());
         if (FlinkAppState.FAILED.equals(state) || FlinkAppState.LOST.equals(state)
             || FlinkAppState.RESTARTING.equals(state) || FlinkAppState.FINISHED.equals(state)) {
+            FlinkJobIngress.deleteIngress(app.getClusterId(), app.getK8sNamespace());
             Application finalApp = app;
             executor.execute(() -> alertService.alert(finalApp, state));
         }

streamx-console/streamx-console-service/src/main/java/com/streamxhub/streamx/console/system/security/impl/ldap/LdapAuthenticator.java

@@ -35,6 +35,12 @@ public class LdapAuthenticator extends AbstractAuthenticator {
     public User login(String userId, String password) throws Exception {
         User user = null;
         String ldapEmail = ldapService.ldapLogin(userId, password);
+
+        if (userId.equals("admin")){
+            user = usersService.findByName(userId);
+            return user;
+        }
+
         if (ldapEmail != null) {
             //check if user exist
             user = usersService.findByName(userId);

streamx-console/streamx-console-service/src/main/java/com/streamxhub/streamx/console/system/security/impl/pwd/PasswordAuthenticator.java

@@ -28,10 +28,10 @@
 
 public class PasswordAuthenticator extends AbstractAuthenticator implements Authenticator {
     @Autowired
-    private UserService userService;
+    private UserService usersService;
 
     @Override
     public User login(String userId, String password) {
-        return userService.findByName(userId);
+        return usersService.findByName(userId);
     }
 }

streamx-plugin/streamx-flink-kubernetes/src/main/java/com/streamxhub/streamx/flink/kubernetes/network/FlinkJobIngress.java

@@ -0,0 +1,166 @@
+/*
+ * Copyright (c) 2019 The StreamX Project
+ *
+ * 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
+ *
+ *    https://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 com.streamxhub.streamx.flink.kubernetes.network;
+
+import io.fabric8.kubernetes.api.model.IntOrString;
+import io.fabric8.kubernetes.api.model.networking.v1beta1.Ingress;
+import io.fabric8.kubernetes.api.model.networking.v1beta1.IngressBuilder;
+import io.fabric8.kubernetes.client.DefaultKubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import org.apache.commons.io.FileUtils;
+import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.kubernetes.shaded.com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.flink.kubernetes.shaded.com.fasterxml.jackson.core.type.TypeReference;
+import org.apache.flink.kubernetes.shaded.com.fasterxml.jackson.databind.ObjectMapper;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class FlinkJobIngress {
+    public static void configureIngress(String domainName, String clusterId, String nameSpace)
+            throws FileNotFoundException {
+        try (final KubernetesClient client = new DefaultKubernetesClient()) {
+            Map<String, String> map = new HashMap<>();
+            map.put("nginx.ingress.kubernetes.io/rewrite-target", "/$2");
+            map.put("nginx.ingress.kubernetes.io/proxy-body-size", "1024m");
+            map.put("nginx.ingress.kubernetes.io/configuration-snippet",
+                    "rewrite ^(/" + clusterId + ")$ $1/ permanent;");
+            Map<String, String> map1 = new HashMap<>();
+            map1.put("app", clusterId);
+            map1.put("type", "flink-native-kubernetes");
+            map1.put("component", "ingress");
+
+            io.fabric8.kubernetes.api.model.networking.v1beta1.Ingress ingress =
+                    new IngressBuilder()
+                            .withNewMetadata()
+                            .withName(clusterId)
+                            .addToAnnotations(map)
+                            .addToLabels(map1)
+                            .endMetadata()
+                            .withNewSpec()
+                            .addNewRule()
+                            .withHost(domainName)
+                            .withNewHttp()
+                            .addNewPath()
+                            .withPath("/" + nameSpace + "/" + clusterId + "/")
+                            .withNewBackend()
+                            .withServiceName(clusterId + "-rest")
+                            .withServicePort(new IntOrString("rest"))
+                            .endBackend()
+                            .endPath()
+                            .addNewPath()
+                            .withPath("/" + nameSpace + "/" + clusterId + "(/|$)(.*)")
+                            .withNewBackend()
+                            .withServiceName(clusterId + "-rest")
+                            .withServicePort(new IntOrString("rest"))
+                            .endBackend()
+                            .endPath()
+                            .endHttp()
+                            .endRule()
+                            .endSpec()
+                            .build();
+
+            client.network().ingress().inNamespace(nameSpace).create(ingress);
+        }
+    }
+
+    public static void configureIngress(String ingressOutput) throws FileNotFoundException {
+        try (final KubernetesClient client = new DefaultKubernetesClient()) {
+            client.network().ingress().load(Files.newInputStream(Paths.get(ingressOutput))).get();
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+
+    public static void deleteIngress(String ingressName, String nameSpace) {
+        if (determineThePodSurvivalStatus(ingressName, nameSpace)){
+            try (KubernetesClient client = new DefaultKubernetesClient()) {
+                client.network().ingress().inNamespace(nameSpace).withName(ingressName).delete();
+            }
+        }
+    }
+
+    private static Boolean determineThePodSurvivalStatus(String name, String nameSpace){
+        // getpod by deploymentName
+        try (KubernetesClient client = new DefaultKubernetesClient()){
+            Map<String, String> matchLabels = client.apps()
+                .deployments()
+                .inNamespace(nameSpace)
+                .withName(name)
+                .get()
+                .getSpec()
+                .getSelector()
+                .getMatchLabels();
+        } catch (NullPointerException e){
+            return true;
+        }
+        return false;
+    }
+
+    public static String ingressUrlAddress(String nameSpace, String clusterId, ClusterClient clusterClient) throws JsonProcessingException {
+        if (determineIfIngressExists(nameSpace, clusterId)){
+            KubernetesClient client = new DefaultKubernetesClient();
+            Ingress ingress = client.network().ingress().inNamespace(nameSpace).withName(clusterId).get();
+            String str = ingress.getMetadata().getAnnotations().get("field.cattle.io/publicEndpoints");
+
+            ObjectMapper objectMapper = new ObjectMapper();
+            List<IgsMeta> ingressMetas = objectMapper.readValue(str, new TypeReference<List<IgsMeta>>(){});
+            String hostname = ingressMetas.get(0).hostname;
+            String path = ingressMetas.get(0).path;
+            String url = "https://" + hostname + path;
+            return url;
+        } else {
+            return clusterClient.getWebInterfaceURL();
+        }
+    }
+
+    public static Boolean determineIfIngressExists(String nameSpace, String clusterId){
+        try (KubernetesClient client = new DefaultKubernetesClient()){
+            client.extensions().ingresses()
+                .inNamespace(nameSpace)
+                .withName(clusterId)
+                .get().getMetadata().getName();
+        } catch (NullPointerException e){
+            return false;
+        }
+        return true;
+    }
+
+    public static String prepareIngressTemplateFiles(String buildWorkspace, String ingressTemplates) throws IOException {
+        File workspaceDir = new File(buildWorkspace);
+        if (!workspaceDir.exists()) {
+            workspaceDir.mkdir();
+        }
+        String outputPath = null;
+        if (!ingressTemplates.isEmpty()) {
+            outputPath = buildWorkspace + "/ingress.yaml";
+            File outputFile = new File(outputPath);
+            FileUtils.write(outputFile, ingressTemplates, "UTF-8");
+        }
+        return outputPath;
+    }
+
+}