集群配置管理

dlink-admin/src/main/java/com/dlink/controller/ClusterConfigurationController.java

+package com.dlink.controller;
+
+import com.dlink.common.result.ProTableResult;
+import com.dlink.common.result.Result;
+import com.dlink.model.ClusterConfiguration;
+import com.dlink.service.ClusterConfigurationService;
+import com.fasterxml.jackson.databind.JsonNode;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.*;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * ClusterConfigController
+ *
+ * @author wenmo
+ * @since 2021/11/6 21:16
+ */
+@Slf4j
+@RestController
+@RequestMapping("/api/clusterConfiguration")
+public class ClusterConfigurationController {
+    @Autowired
+    private ClusterConfigurationService clusterConfigurationService;
+
+    /**
+     * 新增或者更新
+     */
+    @PutMapping
+    public Result saveOrUpdate(@RequestBody ClusterConfiguration clusterConfiguration) {
+        if(clusterConfigurationService.saveOrUpdate(clusterConfiguration)){
+            return Result.succeed("新增成功");
+        }else {
+            return Result.failed("新增失败");
+        }
+    }
+
+    /**
+     * 动态查询列表
+     */
+    @PostMapping
+    public ProTableResult<ClusterConfiguration> listClusterConfigs(@RequestBody JsonNode para) {
+        return clusterConfigurationService.selectForProTable(para);
+    }
+
+    /**
+     * 批量删除
+     */
+    @DeleteMapping
+    public Result deleteMul(@RequestBody JsonNode para) {
+        if (para.size()>0){
+            List<Integer> error = new ArrayList<>();
+            for (final JsonNode item : para){
+                Integer id = item.asInt();
+                if(!clusterConfigurationService.removeById(id)){
+                    error.add(id);
+                }
+            }
+            if(error.size()==0) {
+                return Result.succeed("删除成功");
+            }else {
+                return Result.succeed("删除部分成功，但"+error.toString()+"删除失败，共"+error.size()+"次失败。");
+            }
+        }else{
+            return Result.failed("请选择要删除的记录");
+        }
+    }
+
+    /**
+     * 获取指定ID的信息
+     */
+    @PostMapping("/getOneById")
+    public Result getOneById(@RequestBody ClusterConfiguration clusterConfiguration) {
+        clusterConfiguration = clusterConfigurationService.getById(clusterConfiguration.getId());
+        return Result.succeed(clusterConfiguration,"获取成功");
+    }
+}

dlink-admin/src/main/java/com/dlink/mapper/ClusterConfigurationMapper.java

+package com.dlink.mapper;
+
+import com.dlink.db.mapper.SuperMapper;
+import com.dlink.model.ClusterConfiguration;
+import org.apache.ibatis.annotations.Mapper;
+
+/**
+ * ClusterMapper
+ *
+ * @author wenmo
+ * @since 2021/5/28 13:56
+ **/
+@Mapper
+public interface ClusterConfigurationMapper extends SuperMapper<ClusterConfiguration> {
+}

dlink-admin/src/main/java/com/dlink/model/ClusterConfiguration.java

+package com.dlink.model;
+
+import com.baomidou.mybatisplus.annotation.FieldFill;
+import com.baomidou.mybatisplus.annotation.TableField;
+import com.baomidou.mybatisplus.annotation.TableName;
+import com.dlink.assertion.Asserts;
+import com.dlink.db.model.SuperEntity;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * ClusterConfig
+ *
+ * @author wenmo
+ * @since 2021/11/6
+ **/
+@Data
+@EqualsAndHashCode(callSuper = false)
+@TableName("dlink_cluster_config")
+public class ClusterConfiguration extends SuperEntity {
+
+    private static final long serialVersionUID = 5830130188542066241L;
+
+    @TableField(fill = FieldFill.INSERT)
+    private String alias;
+
+    private String type;
+
+    private String configJson;
+
+    private boolean isAvailable;
+
+    private String note;
+
+    @TableField(exist = false)
+    private Map<String,String> config = new HashMap<>();
+
+
+    public Map<String,String> parseConfig(){
+        ObjectMapper objectMapper = new ObjectMapper();
+        try {
+            if(Asserts.isNotNullString(configJson)) {
+                config = objectMapper.readValue(configJson, HashMap.class);
+            }
+        } catch (JsonProcessingException e) {
+            e.printStackTrace();
+        }
+        return config;
+    }
+}

dlink-admin/src/main/java/com/dlink/service/ClusterConfigurationService.java

+package com.dlink.service;
+
+import com.dlink.db.service.ISuperService;
+import com.dlink.model.ClusterConfiguration;
+
+/**
+ * ClusterConfigService
+ *
+ * @author wenmo
+ * @since 2021/11/6 20:52
+ */
+public interface ClusterConfigurationService extends ISuperService<ClusterConfiguration> {
+
+    ClusterConfiguration getClusterConfigById(Integer id);
+}

dlink-admin/src/main/java/com/dlink/service/impl/ClusterConfigurationServiceImpl.java

+package com.dlink.service.impl;
+
+import com.dlink.db.service.impl.SuperServiceImpl;
+import com.dlink.mapper.ClusterConfigurationMapper;
+import com.dlink.model.ClusterConfiguration;
+import com.dlink.service.ClusterConfigurationService;
+import org.springframework.stereotype.Service;
+
+/**
+ * ClusterConfigServiceImpl
+ *
+ * @author wenmo
+ * @since 2021/11/6 20:54
+ */
+@Service
+public class ClusterConfigurationServiceImpl extends SuperServiceImpl<ClusterConfigurationMapper,ClusterConfiguration> implements ClusterConfigurationService {
+    @Override
+    public ClusterConfiguration getClusterConfigById(Integer id) {
+        ClusterConfiguration clusterConfiguration = baseMapper.selectById(id);
+        clusterConfiguration.parseConfig();
+        return clusterConfiguration;
+    }
+}

dlink-admin/src/main/resources/mapper/ClusterConfigurationMapper.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="com.dlink.mapper.ClusterConfigurationMapper">
+
+    <!-- 通用查询映射结果 -->
+    <resultMap id="BaseResultMap" type="com.dlink.model.ClusterConfiguration">
+        <id column="id" property="id" />
+        <result column="name" property="name" />
+        <result column="alias" property="alias" />
+        <result column="type" property="type" />
+        <result column="config_json" property="configJson" />
+        <result column="is_available" property="isAvailable" />
+        <result column="note" property="note" />
+        <result column="enabled" property="enabled" />
+        <result column="create_time" property="createTime" />
+        <result column="update_time" property="updateTime" />
+    </resultMap>
+
+    <!-- 通用查询结果列 -->
+    <sql id="Base_Column_List">
+        id, name, alias, type,config_json,is_available,note, enabled, create_time, update_time
+    </sql>
+
+
+    <select id="selectForProTable" resultType="com.dlink.model.ClusterConfiguration">
+        select
+        a.*
+        from
+        dlink_cluster_configuration a
+        <where>
+            1=1
+            <if test='param.name!=null and param.name!=""'>
+                and a.name like "%${param.name}%"
+            </if>
+            <if test='param.alias!=null and param.alias!=""'>
+                and a.alias like "%${param.alias}%"
+            </if>
+            <if test='param.createTime!=null and param.createTime!=""'>
+                and a.create_time <![CDATA[>=]]> str_to_date( #{param.createTime},'%Y-%m-%d %H:%i:%s')
+            </if>
+            <if test='param.updateTime!=null and param.updateTime!=""'>
+                and a.update_time <![CDATA[>=]]> str_to_date( #{param.updateTime},'%Y-%m-%d %H:%i:%s')
+            </if>
+            <if test='ew.sqlSegment!=null and ew.sqlSegment!="" and !ew.sqlSegment.startsWith(" ORDER BY")'>
+                and
+            </if>
+            <if test='ew.sqlSegment!=null and ew.sqlSegment!=""'>
+                ${ew.sqlSegment}
+            </if>
+        </where>
+    </select>
+</mapper>

dlink-extends/dependency-reduced-pom.xml

@@ -60,7 +60,7 @@
     </plugins>
   </build>
   <properties>
-    <flink.version>1.12.5</flink.version>
+    <flink.version>1.13.3</flink.version>
     <java.version>1.8</java.version>
     <clickhouse-jdbc.version>0.2.6</clickhouse-jdbc.version>
     <mariadb-java-client.version>2.5.4</mariadb-java-client.version>

dlink-gateway/src/main/java/com/dlink/gateway/yarn/YarnGateway.java

@@ -26,6 +26,7 @@ import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.client.api.YarnClient;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 
+import java.net.URI;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -69,7 +70,9 @@ public abstract class YarnGateway extends AbstractGateway {
 
     private void initYarnClient(){
         yarnConfiguration = new YarnConfiguration();
-        yarnConfiguration.addResource( new Path( config.getClusterConfig().getYarnConfigPath() ) );
+        yarnConfiguration.addResource( new Path( URI.create(config.getClusterConfig().getYarnConfigPath()+"/yarn-site.xml") ) );
+        yarnConfiguration.addResource( new Path( URI.create(config.getClusterConfig().getYarnConfigPath()+"/core-site.xml") ) );
+        yarnConfiguration.addResource( new Path( URI.create(config.getClusterConfig().getYarnConfigPath()+"/hdfs-site.xml") ) );
         yarnClient = YarnClient.createYarnClient();
         yarnClient.init(yarnConfiguration);
         yarnClient.start();

dlink-web/src/pages/ClusterConfiguration/data.d.ts

+export type ClusterConfigerationTableListItem = {
+  id: number,
+  name: string,
+  alias: string,
+  type: string,
+  config: any,
+  isAvailable: boolean,
+  note: string,
+  enabled: boolean,
+  createTime: Date,
+  updateTime: Date,
+};