[1.9.0]Add synchronized department user table scheduled tasks

linkis-extensions/linkis-et-monitor/src/main/java/org/apache/linkis/monitor/department/dao/UserDepartmentInfoMapper.java

@@ -0,0 +1,41 @@
+/*
+ * 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.linkis.monitor.department.dao;
+
+import org.apache.linkis.monitor.department.entity.UserDepartmentInfo;
+
+import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
+
+import java.util.List;
+
+@Mapper
+public interface UserDepartmentInfoMapper {
+
+  void insertUser(UserDepartmentInfo user);
+
+  int batchInsertUsers(@Param("userDepartmentInfos") List<UserDepartmentInfo> userDepartmentInfos);
+
+  void updateUser(UserDepartmentInfo user);
+
+  UserDepartmentInfo selectUser(@Param("userName") String userName);
+
+  void deleteUser();
+
+  List<UserDepartmentInfo> selectAllUsers();
+}

linkis-extensions/linkis-et-monitor/src/main/java/org/apache/linkis/monitor/department/entity/UserDepartmentInfo.java

@@ -0,0 +1,150 @@
+/*
+ * 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.linkis.monitor.department.entity;
+
+import java.util.Date;
+
+public class UserDepartmentInfo {
+
+  private String clusterCode;
+
+  private String userType;
+  private String userName;
+  private String orgId;
+  private String orgName;
+  private String queueName;
+  private String dbName;
+  private String interfaceUser;
+  private String isUnionAnalyse;
+  private Date createTime;
+  private String userItsmNo;
+
+  // 构造函数、getter和setter方法
+  public UserDepartmentInfo(
+      String clusterCode,
+      String userType,
+      String userName,
+      String orgId,
+      String orgName,
+      String queueName,
+      String dbName,
+      String interfaceUser,
+      String isUnionAnalyse,
+      Date createTime,
+      String userItsmNo) {
+    this.clusterCode = clusterCode;
+    this.userType = userType;
+    this.userName = userName;
+    this.orgId = orgId;
+    this.orgName = orgName;
+    this.queueName = queueName;
+    this.dbName = dbName;
+    this.interfaceUser = interfaceUser;
+    this.isUnionAnalyse = isUnionAnalyse;
+    this.createTime = createTime;
+    this.userItsmNo = userItsmNo;
+  }
+
+  public String getClusterCode() {
+    return clusterCode;
+  }
+
+  public void setClusterCode(String clusterCode) {
+    this.clusterCode = clusterCode;
+  }
+
+  public String getUserType() {
+    return userType;
+  }
+
+  public void setUserType(String userType) {
+    this.userType = userType;
+  }
+
+  public String getUserName() {
+    return userName;
+  }
+
+  public void setUserName(String userName) {
+    this.userName = userName;
+  }
+
+  public String getOrgId() {
+    return orgId;
+  }
+
+  public void setOrgId(String orgId) {
+    this.orgId = orgId;
+  }
+
+  public String getOrgName() {
+    return orgName;
+  }
+
+  public void setOrgName(String orgName) {
+    this.orgName = orgName;
+  }
+
+  public String getQueueName() {
+    return queueName;
+  }
+
+  public void setQueueName(String queueName) {
+    this.queueName = queueName;
+  }
+
+  public String getDbName() {
+    return dbName;
+  }
+
+  public void setDbName(String dbName) {
+    this.dbName = dbName;
+  }
+
+  public String getInterfaceUser() {
+    return interfaceUser;
+  }
+
+  public void setInterfaceUser(String interfaceUser) {
+    this.interfaceUser = interfaceUser;
+  }
+
+  public String getIsUnionAnalyse() {
+    return isUnionAnalyse;
+  }
+
+  public void setIsUnionAnalyse(String isUnionAnalyse) {
+    this.isUnionAnalyse = isUnionAnalyse;
+  }
+
+  public Date getCreateTime() {
+    return createTime;
+  }
+
+  public void setCreateTime(Date createTime) {
+    this.createTime = createTime;
+  }
+
+  public String getUserItsmNo() {
+    return userItsmNo;
+  }
+
+  public void setUserItsmNo(String userItsmNo) {
+    this.userItsmNo = userItsmNo;
+  }
+}

linkis-extensions/linkis-et-monitor/src/main/java/org/apache/linkis/monitor/scheduled/UserDepartmentInfoSync.java

@@ -0,0 +1,103 @@
+/*
+ * 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.linkis.monitor.scheduled;
+
+import org.apache.linkis.monitor.constants.Constants;
+import org.apache.linkis.monitor.department.dao.UserDepartmentInfoMapper;
+import org.apache.linkis.monitor.department.entity.UserDepartmentInfo;
+import org.apache.linkis.monitor.factory.MapperFactory;
+import org.apache.linkis.monitor.utils.alert.AlertDesc;
+import org.apache.linkis.monitor.utils.alert.ims.MonitorAlertUtils;
+import org.apache.linkis.monitor.utils.alert.ims.PooledImsAlertUtils;
+
+import org.apache.commons.lang3.StringUtils;
+
+import org.springframework.context.annotation.PropertySource;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import org.springframework.util.CollectionUtils;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@Component
+@PropertySource(value = "classpath:linkis-et-monitor.properties", encoding = "UTF-8")
+public class UserDepartmentInfoSync {
+
+  private static final Logger logger = LoggerFactory.getLogger(ResourceMonitor.class);
+
+  @Scheduled(cron = "${linkis.monitor.org.user.sync.cron:0 0 0 0/7 * ?}")
+  public static void DepartmentInfoSync() {
+
+    UserDepartmentInfoMapper userDepartmentInfoMapper = MapperFactory.getUserDepartmentInfoMapper();
+    // 获取linkis_org_user_sync信息
+    List<UserDepartmentInfo> userDepartmentInfos = userDepartmentInfoMapper.selectAllUsers();
+
+    if (CollectionUtils.isEmpty(userDepartmentInfos)) {
+      logger.info("No user department info to sync");
+      // 并且发送告警通知
+      return;
+    } else {
+      logger.info("Start to sync user department info");
+
+      List<UserDepartmentInfo> alterList =
+          userDepartmentInfos.stream()
+              .filter(
+                  userDepartmentInfo ->
+                      StringUtils.isNotBlank(userDepartmentInfo.getUserName())
+                          && (StringUtils.isBlank(userDepartmentInfo.getOrgId())
+                              || StringUtils.isBlank(userDepartmentInfo.getOrgName())))
+              .collect(Collectors.toList());
+      List<UserDepartmentInfo> syncList =
+          userDepartmentInfos.stream()
+              .filter(
+                  userDepartmentInfo ->
+                      StringUtils.isNotBlank(userDepartmentInfo.getUserName())
+                          && StringUtils.isNotBlank(userDepartmentInfo.getOrgId())
+                          && StringUtils.isNotBlank(userDepartmentInfo.getOrgName()))
+              .collect(Collectors.toList());
+      if (!CollectionUtils.isEmpty(alterList)) {
+        // 统计异常名称，然后发送告警
+        String usernames =
+            alterList.stream()
+                .filter(s -> StringUtils.isNotBlank(s.getUserName()))
+                .map(UserDepartmentInfo::getUserName)
+                .limit(5)
+                .collect(Collectors.joining(","));
+        if (StringUtils.isNotBlank(usernames)) {
+          HashMap<String, String> replaceParm = new HashMap<>();
+          replaceParm.put("$user", usernames);
+          replaceParm.put("$count", String.valueOf(alterList.size()));
+          Map<String, AlertDesc> ecmResourceAlerts =
+              MonitorAlertUtils.getAlerts(Constants.DEPARTMENT_USER_IM(), replaceParm);
+          PooledImsAlertUtils.addAlert(ecmResourceAlerts.get("12019"));
+        }
+      }
+      if (!CollectionUtils.isEmpty(syncList)) {
+        // 删除org_user数据，再同步
+        userDepartmentInfoMapper.deleteUser();
+        userDepartmentInfoMapper.batchInsertUsers(syncList);
+      }
+    }
+  }
+}

linkis-extensions/linkis-et-monitor/src/main/resources/mapper/common/UserDepartmentInfoMapper.xml

@@ -0,0 +1,57 @@
+<?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="org.apache.linkis.monitor.department.dao.UserDepartmentInfoMapper">
+
+    <!-- 新增方法 -->
+    <insert id="insertUser">
+        INSERT INTO linkis_org_user(cluster_code, user_type, user_name, org_id, org_name, queue_name, db_name, interface_user, is_union_analyse, create_time, user_itsm_no)
+        VALUES (#{clusterCode}, #{userType}, #{userName}, #{orgId}, #{orgName}, #{queueName}, #{dbName}, #{interfaceUser}, #{isUnionAnalyse}, #{createTime}, #{userItsmNo})
+    </insert>
+
+    <!-- 批量新增方法 -->
+    <insert id="batchInsertUsers" parameterType="java.util.List">
+        INSERT INTO linkis_org_user (cluster_code, user_type, user_name, org_id, org_name, queue_name, db_name, interface_user, is_union_analyse, create_time, user_itsm_no)
+        VALUES
+        <foreach item="user" index="index" collection="userDepartmentInfos" separator=",">
+            (#{user.clusterCode}, #{user.userType}, #{user.userName}, #{user.orgId}, #{user.orgName}, #{user.queueName}, #{user.dbName}, #{user.interfaceUser}, #{user.isUnionAnalyse}, #{user.createTime}, #{user.userItsmNo})
+        </foreach>
+    </insert>
+
+    <!-- 修改方法 -->
+    <update id="updateUser">
+        UPDATE linkis_org_user
+        <set>
+            <if test="clusterCode != null">cluster_code = #{clusterCode},</if>
+            <if test="userType != null">user_type = #{userType},</if>
+            <if test="orgId != null">org_id = #{orgId},</if>
+            <if test="orgName != null">org_name = #{orgName},</if>
+            <if test="queueName != null">queue_name = #{queueName},</if>
+            <if test="dbName != null">db_name = #{dbName},</if>
+            <if test="interfaceUser != null">interface_user = #{interfaceUser},</if>
+            <if test="isUnionAnalyse != null">is_union_analyse = #{isUnionAnalyse},</if>
+            <if test="createTime != null">create_time = #{createTime},</if>
+            <if test="userItsmNo != null">user_itsm_no = #{userItsmNo},</if>
+        </set>
+        WHERE user_name = #{userName}
+    </update>
+
+    <!-- 查询方法 -->
+    <select id="selectUser" resultType="org.apache.linkis.monitor.department.entity.UserDepartmentInfo">
+        SELECT * FROM linkis_org_user WHERE user_name = #{userName}
+    </select>
+
+    <!-- 删除方法 -->
+    <delete id="deleteUser">
+        DELETE FROM linkis_org_user
+    </delete>
+
+    <!-- 查询所有数据 -->
+    <select id="selectAllUsers" resultType="org.apache.linkis.monitor.department.entity.UserDepartmentInfo">
+        SELECT * FROM linkis_org_user_sync
+    </select>
+
+
+
+
+</mapper>

linkis-extensions/linkis-et-monitor/src/main/scala/org/apache/linkis/monitor/LinkisJobHistoryScanSpringConfiguration.scala

@@ -17,6 +17,7 @@
 
 package org.apache.linkis.monitor
 
+import org.apache.linkis.monitor.department.dao.UserDepartmentInfoMapper
 import org.apache.linkis.monitor.factory.MapperFactory
 import org.apache.linkis.monitor.instance.dao.InstanceInfoDao
 import org.apache.linkis.monitor.jobhistory.dao.JobHistoryMapper
@@ -39,10 +40,14 @@ class LinkisJobHistoryScanSpringConfiguration {
   @Autowired
   private var instanceInfoMapper: InstanceInfoDao = _
 
+  @Autowired
+  private var userDepartmentInfoMapper: UserDepartmentInfoMapper = _
+
   @PostConstruct
   def init(): Unit = {
     MapperFactory.setJobHistoryMapper(jobHistoryMapper)
     MapperFactory.setInstanceInfoMapper(instanceInfoMapper)
+    MapperFactory.setUserDepartmentInfoMapper(userDepartmentInfoMapper)
   }
 
 }

linkis-extensions/linkis-et-monitor/src/main/scala/org/apache/linkis/monitor/constants/Constants.scala

@@ -79,6 +79,7 @@ object Constants {
   val BML_CLEAR_IM = "bml.clear.monitor.im."
   val THREAD_TIME_OUT_IM = "thread.monitor.timeout.im."
   val JOB_RESULT_IM = "jobhistory.result.monitor.im."
+  val DEPARTMENT_USER_IM = "department.user.sync.im."
 
   val BML_VERSION_MAX_NUM: CommonVars[Int] =
     CommonVars[Int]("linkis.monitor.bml.cleaner.version.max.num", 50)