liuyupeng/saas-houduan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1、mtr探测设备增加并发限制。

Browse Source 
2、开发防火墙策略区分ipv4和ipv6。
3、脚本执行策略下发时间优化。
This commit is contained in:
gaoyutao
2025-12-03 19:18:12 +08:00
parent a40a6f344e
commit a5b5f5b08d
18 changed files with 249 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tongran-modules/tongran-mtragent/src/main/java/com/tongran/mtragent/controller/RmMtrPolicyConfigController.java
+2 -1
View File
@@ -76,7 +76,8 @@ public class RmMtrPolicyConfigController extends BaseController
@PostMapping
public AjaxResult add(@RequestBody RmMtrPolicyConfig rmMtrPolicyConfig)
{
return toAjax(rmMtrPolicyConfigService.insertRmMtrPolicyConfig(rmMtrPolicyConfig));
int rows = rmMtrPolicyConfigService.insertRmMtrPolicyConfig(rmMtrPolicyConfig);
return toAjax(rows);
}
/**
tongran-modules/tongran-mtragent/src/main/java/com/tongran/mtragent/domain/InitialHeartbeatListen.java
+2
View File
@@ -33,5 +33,7 @@ public class InitialHeartbeatListen extends BaseEntity
private String version;
/** 服务启动时间 */
private Long startupTime;
/** cpu核数 */
private Integer cpucores;
}
tongran-modules/tongran-mtragent/src/main/java/com/tongran/mtragent/domain/RmMtrClientRegistration.java
+2
View File
@@ -91,4 +91,6 @@ public class RmMtrClientRegistration extends BaseEntity
private String mtrClientIds;
/** 多条件 */
private String queryName;
/** cpu核数 */
private Integer cpucores;
}
tongran-modules/tongran-mtragent/src/main/java/com/tongran/mtragent/handler/MessageHandler.java
+5
View File
@@ -371,6 +371,11 @@ public class MessageHandler {
}
needUpdate = true;
}
if(rmMtrClientRegistration.getCpucores() == null ||
rmMtrClientRegistration.getCpucores() != heartbeat.getCpucores()){
updateData.setCpucores(heartbeat.getCpucores());
needUpdate = true;
}
if(needUpdate){
rmMtrClientRegistrationService.updateRmMtrClientRegistration(updateData);
}
tongran-modules/tongran-mtragent/src/main/java/com/tongran/mtragent/service/impl/RmMtrClientRegistrationServiceImpl.java
+2
View File
@@ -65,6 +65,8 @@ public class RmMtrClientRegistrationServiceImpl implements IRmMtrClientRegistrat
@Override
public List<RmMtrClientRegistration> selectRmMtrClientRegistrationList(RmMtrClientRegistration rmMtrClientRegistration)
{
rmMtrClientRegistration.setOnlineStatus("1");
rmMtrClientRegistration.setRegisterStatus("1");
List<RmMtrClientRegistration> list = rmMtrClientRegistrationMapper.selectRmMtrClientRegistrationList(rmMtrClientRegistration);
for (RmMtrClientRegistration mtrClientRegistration : list) {
// 处理网卡信息
tongran-modules/tongran-mtragent/src/main/java/com/tongran/mtragent/service/impl/RmMtrPolicyConfigServiceImpl.java
+78 -2
View File
@@ -5,7 +5,9 @@ import com.tongran.common.core.domain.R;
import com.tongran.common.core.utils.DateUtils;
import com.tongran.common.core.utils.StringUtils;
import com.tongran.common.security.utils.SecurityUtils;
import com.tongran.mtragent.domain.RmMtrClientRegistration;
import com.tongran.mtragent.domain.RmMtrPolicyConfig;
import com.tongran.mtragent.mapper.RmMtrClientRegistrationMapper;
import com.tongran.mtragent.mapper.RmMtrPolicyConfigMapper;
import com.tongran.mtragent.service.IRmMtrPolicyConfigService;
import com.tongran.system.api.RemoteRocketMqService;
@@ -28,6 +30,8 @@ public class RmMtrPolicyConfigServiceImpl implements IRmMtrPolicyConfigService
@Autowired
private RmMtrPolicyConfigMapper rmMtrPolicyConfigMapper;
@Autowired
private RmMtrClientRegistrationMapper rmMtrClientRegistrationMapper;
@Autowired
private RemoteRocketMqService remoteRocketMqService;
/**
@@ -122,13 +126,15 @@ public class RmMtrPolicyConfigServiceImpl implements IRmMtrPolicyConfigService
}
/**
* 新增mtr探测策略配置
*
*
* @param rmMtrPolicyConfig mtr探测策略配置
* @return 结果
*/
@Override
public int insertRmMtrPolicyConfig(RmMtrPolicyConfig rmMtrPolicyConfig)
{
// 检查是否超出限制
checkServerNum(rmMtrPolicyConfig);
rmMtrPolicyConfig.setCreateTime(DateUtils.getNowDate());
rmMtrPolicyConfig.setUpdateTime(DateUtils.getNowDate());
rmMtrPolicyConfig.setCreateBy(SecurityUtils.getUsername());
@@ -142,19 +148,89 @@ public class RmMtrPolicyConfigServiceImpl implements IRmMtrPolicyConfigService
/**
* 修改mtr探测策略配置
*
*
* @param rmMtrPolicyConfig mtr探测策略配置
* @return 结果
*/
@Override
public int updateRmMtrPolicyConfig(RmMtrPolicyConfig rmMtrPolicyConfig)
{
// 检查是否超出限制
checkServerNum(rmMtrPolicyConfig);
rmMtrPolicyConfig.setUpdateTime(DateUtils.getNowDate());
// 给ip赋值
setServerip(rmMtrPolicyConfig);
return rmMtrPolicyConfigMapper.updateRmMtrPolicyConfig(rmMtrPolicyConfig);
}
public void checkServerNum(RmMtrPolicyConfig rmMtrPolicyConfig){
if (rmMtrPolicyConfig.getMtrClientId() == null) {
throw new RuntimeException("MTR客户端ID不能为空");
}
// 获取CPU核心数
int cpuCores = getCpuCores(rmMtrPolicyConfig.getMtrClientId());
if(cpuCores == 0){
// cpu核数未采集
throw new RuntimeException("该MTRAgent的cpu核数未上报，请检查详情，待cpu核数上报后重试");
}
Set<String> serverSet = isPolicyCountExceeded(rmMtrPolicyConfig);
if(serverSet == null) {
return; // 没有服务器或客户端ID为空，直接返回
}
if(serverSet.size() > cpuCores){
// 设置错误信息，包含已下发的服务器列表
String errorMessage = String.format("探测服务器数量超过节点并发数量限制(%d)，已开启探测的服务器clientId为: %s",
cpuCores, String.join(", ", serverSet));
throw new RuntimeException(errorMessage);
}
}
/**
* 检查策略数量是否超过CPU核心数限制
*
* @param newPolicy 新增或修改的策略
* @return true-超过限制，false-未超过限制
*/
private Set<String> isPolicyCountExceeded(RmMtrPolicyConfig newPolicy) {
// 获取当前客户端应该执行的策略列表
List<RmMtrPolicyConfig> currentPolicies = getPoliciesForMtrClient(newPolicy.getMtrClientId());
// 获取新策略中的服务器ID列表（去重）
List<String> newServerIds = getServerIdsFromPolicy(newPolicy);
if (newServerIds.isEmpty()) {
return null;
}
// 获取当前已分配的服务器ID列表（去重）
Set<String> currentServerIds = new HashSet<>();
for (RmMtrPolicyConfig policy : currentPolicies) {
// 如果是修改操作，需要排除当前策略本身
if (newPolicy.getId() != null &&
newPolicy.getId().equals(policy.getId())) {
continue; // 跳过当前正在修改的策略
}
List<String> serverIds = getServerIdsFromPolicy(policy);
currentServerIds.addAll(serverIds);
}
// 计算新增的不重复服务器数量
Set<String> allServerIds = new HashSet<>(currentServerIds);
allServerIds.addAll(newServerIds);
return allServerIds;
}
/**
* 获取CPU核心数
*/
private int getCpuCores(String mtrClientId) {
RmMtrClientRegistration query = new RmMtrClientRegistration();
query.setMtrClientId(mtrClientId);
RmMtrClientRegistration rmMtrRegistMsg = rmMtrClientRegistrationMapper.getMsgByMtrClientId(query);
if(rmMtrRegistMsg != null && rmMtrRegistMsg.getCpucores() != null){
return rmMtrRegistMsg.getCpucores() * 2;
}
return 0;
}
/**
* 批量删除mtr探测策略配置
*
tongran-modules/tongran-mtragent/src/main/resources/mapper/mtragent/RmMtrClientRegistrationMapper.xml
+8 -4
View File
@@ -3,7 +3,7 @@
PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
"http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.tongran.mtragent.mapper.RmMtrClientRegistrationMapper">
<resultMap type="RmMtrClientRegistration" id="RmMtrClientRegistrationResult">
<result property="id" column="id" />
<result property="mtrClientId" column="mtr_client_id" />
@@ -13,6 +13,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
<result property="registerTime" column="register_time" />
<result property="registerStatus" column="register_status" />
<result property="onlineStatus" column="online_status" />
<result property="cpucores" column="cpucores" />
<result property="heartbeatInterval" column="heartbeat_interval" />
<result property="heartbeatCount" column="heartbeat_count" />
<result property="method" column="method" />
@@ -29,7 +30,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
</resultMap>
<sql id="selectRmMtrClientRegistrationVo">
select id, mtr_client_id, description, version, logical_node, register_time, register_status, online_status, heartbeat_interval, heartbeat_count, method, scheduled_update_time, file_path, file_md5, last_update_result, last_update_time, network_info, create_time, update_time, create_by, update_by from rm_mtr_client_registration
select id, mtr_client_id, description, version, logical_node, register_time, register_status, online_status, cpucores, heartbeat_interval, heartbeat_count, method, scheduled_update_time, file_path, file_md5, last_update_result, last_update_time, network_info, create_time, update_time, create_by, update_by from rm_mtr_client_registration
</sql>
<select id="selectRmMtrClientRegistrationList" parameterType="RmMtrClientRegistration" resultMap="RmMtrClientRegistrationResult">
@@ -80,6 +81,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
<if test="registerTime != null">register_time,</if>
<if test="registerStatus != null and registerStatus != ''">register_status,</if>
<if test="onlineStatus != null and onlineStatus != ''">online_status,</if>
<if test="cpucores != null">cpucores,</if>
<if test="heartbeatInterval != null">heartbeat_interval,</if>
<if test="heartbeatCount != null">heartbeat_count,</if>
<if test="method != null">method,</if>
@@ -93,7 +95,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
<if test="updateTime != null">update_time,</if>
<if test="createBy != null">create_by,</if>
<if test="updateBy != null">update_by,</if>
</trim>
</trim>
<trim prefix="values (" suffix=")" suffixOverrides=",">
<if test="mtrClientId != null and mtrClientId != ''">#{mtrClientId},</if>
<if test="description != null">#{description},</if>
@@ -102,6 +104,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
<if test="registerTime != null">#{registerTime},</if>
<if test="registerStatus != null and registerStatus != ''">#{registerStatus},</if>
<if test="onlineStatus != null and onlineStatus != ''">#{onlineStatus},</if>
<if test="cpucores != null">#{cpucores},</if>
<if test="heartbeatInterval != null">#{heartbeatInterval},</if>
<if test="heartbeatCount != null">#{heartbeatCount},</if>
<if test="method != null">#{method},</if>
@@ -115,7 +118,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
<if test="updateTime != null">#{updateTime},</if>
<if test="createBy != null">#{createBy},</if>
<if test="updateBy != null">#{updateBy},</if>
</trim>
</trim>
</insert>
<update id="updateRmMtrClientRegistration" parameterType="RmMtrClientRegistration">
@@ -128,6 +131,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
<if test="registerTime != null">register_time = #{registerTime},</if>
<if test="registerStatus != null and registerStatus != ''">register_status = #{registerStatus},</if>
<if test="onlineStatus != null and onlineStatus != ''">online_status = #{onlineStatus},</if>
<if test="cpucores != null">cpucores = #{cpucores},</if>
<if test="heartbeatInterval != null">heartbeat_interval = #{heartbeatInterval},</if>
<if test="heartbeatCount != null">heartbeat_count = #{heartbeatCount},</if>
<if test="method != null">method = #{method},</if>
tongran-modules/tongran-system/src/main/java/com/tongran/system/domain/vo/RspVo.java
+2
View File
@@ -12,6 +12,8 @@ public class RspVo {
private String resMag;
/** 路由 */
private String addRoute;
/** 业务网卡名称 */
private String netName;
/** 时间戳 */
private long timestamp = Instant.now().getEpochSecond();
}
tongran-modules/tongran-system/src/main/java/com/tongran/system/service/impl/RmResourceRegistrationServiceImpl.java
+42 -23
View File
@@ -628,6 +628,7 @@ public class RmResourceRegistrationServiceImpl implements IRmResourceRegistratio
rspVo.setResCode(1);
rspVo.setResMag("注册成功");
rspVo.setAddRoute(JSONObject.toJSONString(routeMsg));
rspVo.setNetName(networkInfo.getName());
messageVo.setData(JSONObject.toJSONString(rspVo));
remoteRocketMqService.sendAsyncProducerMessage(
"tr_agent_down", "", "regist_rsp", JSONObject.toJSONString(messageVo), SecurityConstants.INNER
@@ -745,36 +746,54 @@ public class RmResourceRegistrationServiceImpl implements IRmResourceRegistratio
RmNetworkInterfaceRemote queryParam = new RmNetworkInterfaceRemote();
queryParam.setClientId(clientId);
Optional.ofNullable(remoteRocketMqService.getNetworkInterfaceList(queryParam, SecurityConstants.INNER))
R<List<RmNetworkInterfaceRemote>> netList = remoteRocketMqService.getNetworkInterfaceList(queryParam, SecurityConstants.INNER);
// 获取所有bindIp为1或3的网络接口
List<RmNetworkInterfaceRemote> filteredNetworks = Optional.ofNullable(netList)
.map(R::getData)
.flatMap(list -> list.stream()
.map(list -> list.stream()
.filter(ni -> "1".equals(ni.getBindIp()) || "3".equals(ni.getBindIp()))
.findFirst())
.ifPresent(network -> {
updateData.setRegistrationStatus("1");
.collect(Collectors.toList()))
.orElse(Collections.emptyList());
// 发送注册响应
RouteMsg routeMsg = new RouteMsg();
routeMsg.setName(network.getInterfaceName());
routeMsg.setGateway(network.getGateway());
// 如果有符合条件的网络接口
if (!filteredNetworks.isEmpty()) {
updateData.setRegistrationStatus("1");
MessageVo messageVo = new MessageVo();
messageVo.setClientId(clientId);
messageVo.setDataType(MsgEnum.注册应答.getValue());
// 拼接所有符合条件的网络接口名称，用分号隔开
StringBuilder netNameBuilder = new StringBuilder();
for (int i = 0; i < filteredNetworks.size(); i++) {
if (i > 0) {
netNameBuilder.append(";");
}
netNameBuilder.append(filteredNetworks.get(i).getInterfaceName());
}
String netName = netNameBuilder.toString();
RspVo rspVo = new RspVo();
rspVo.setResCode(1);
rspVo.setResMag("注册成功");
rspVo.setAddRoute(JSONObject.toJSONString(routeMsg));
// 发送注册响应
RouteMsg routeMsg = new RouteMsg();
routeMsg.setName(filteredNetworks.get(0).getInterfaceName()); // 使用拼接后的网络接口名称
routeMsg.setGateway(filteredNetworks.get(0).getGateway()); // 使用第一个网络接口的网关，或者您可以根据需要调整
messageVo.setData(JSONObject.toJSONString(rspVo));
MessageVo messageVo = new MessageVo();
messageVo.setClientId(clientId);
messageVo.setDataType(MsgEnum.注册应答.getValue());
remoteRocketMqService.sendAsyncProducerMessage(
"tr_agent_down", "", "regist_rsp", JSONObject.toJSONString(messageVo), SecurityConstants.INNER
);
// 注册成功，下发优先级为0的策略
remoteRocketMqService.issueDefaultPolicyByClientId(clientId, SecurityConstants.INNER);
});
RspVo rspVo = new RspVo();
rspVo.setResCode(1);
rspVo.setResMag("注册成功");
rspVo.setAddRoute(JSONObject.toJSONString(routeMsg));
rspVo.setNetName(netName);
messageVo.setData(JSONObject.toJSONString(rspVo));
remoteRocketMqService.sendAsyncProducerMessage(
"tr_agent_down", "", "regist_rsp", JSONObject.toJSONString(messageVo), SecurityConstants.INNER
);
// 注册成功，下发优先级为0的策略
remoteRocketMqService.issueDefaultPolicyByClientId(clientId, SecurityConstants.INNER);
}
}
rmResourceRegistrationMapper.updateRmResourceRegistration(updateData);