update code comment and doc, bugfix concurrent exception (#1378)

Signed-off-by: tomsun28 <tomsun28@outlook.com>
This commit is contained in:
tomsun28
2023-12-04 14:05:04 +08:00
committed by GitHub
parent c371f8136b
commit 6279fca3fe
357 changed files with 1935 additions and 2666 deletions
@@ -23,9 +23,7 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
/**
* alerter config
*
*
* alerter prop config
*
*/
@Component
@@ -35,7 +33,6 @@ import org.springframework.stereotype.Component;
public class AlerterProperties {
/**
* 告警内容控制台链接
* Alarm content console link
*/
private String consoleUrl = "https://console.tancloud.cn";
@@ -71,41 +68,19 @@ public class AlerterProperties {
private String serverChanNotifyUrl = "https://sctapi.ftqq.com/%s.send";
/**
* 告警评估时间间隔起始基数 每下一次乘2 单位毫秒
* base of alert eval interval time, unit:ms. The next time is 2 times the previous time.
*/
@Deprecated
private long alertEvalIntervalBase = 1000 * 60 * 10L;
/**
* 最大告警评估时间间隔 单位毫秒
* max of alert eval interval time, unit:ms
*/
@Deprecated
private long maxAlertEvalInterval = 1000 * 60 * 60 * 24L;
/**
* 系统内置告警(available alert, reachable alert...)触发次数
* system alert(available alert, reachable alert...) trigger times
*/
@Deprecated
private int systemAlertTriggerTimes = 1;
/**
* Data entry configuration properties 数据入口配置属性
* Data entry configuration properties
*/
private EntranceProperties entrance;
/**
* Data entry configuration properties 数据入口配置属性
* The entry can obtain data from messaging middleware such as kafka rabbitmq rocketmq 入口可以是从kafka rabbitmq rocketmq等消息中间件获取数据
* Data entry configuration properties
*/
@Getter
@Setter
public static class EntranceProperties {
/**
* kafka configuration information kafka配置信息
* kafka configuration information
*/
private KafkaProperties kafka;
@@ -113,20 +88,20 @@ public class AlerterProperties {
@Setter
public static class KafkaProperties {
/**
* Whether the kafka data entry is started kafka数据入口是否启动
* Whether the kafka data entry is started
*/
private boolean enabled = true;
/**
* kafka's connection server url kafka的连接服务器url
* kafka's connection server url
*/
private String servers = "127.0.0.1:9092";
/**
* The name of the topic that receives the data 接收数据的topic名称
* The name of the topic that receives the data
*/
private String topic;
/**
* Consumer Group ID 消费者组ID
* Consumer Group ID
*/
private String groupId;
@@ -28,8 +28,7 @@ import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
* warehouse 工作线程池
*
* alarm module thread pool
*
*/
@Component
@@ -43,11 +42,11 @@ public class AlerterWorkerPool {
}
private void initWorkExecutor() {
// 线程工厂
ThreadFactory threadFactory = new ThreadFactoryBuilder()
.setUncaughtExceptionHandler((thread, throwable) -> {
log.error("workerExecutor has uncaughtException.");
log.error(throwable.getMessage(), throwable); })
log.error(throwable.getMessage(), throwable);
})
.setDaemon(true)
.setNameFormat("alerter-worker-%d")
.build();
@@ -61,9 +60,9 @@ public class AlerterWorkerPool {
}
/**
* Run the alerter task 运行alerter任务
* @param runnable task 任务
* @throws RejectedExecutionException when The thread pool is full of 线程池满
* Run the alerter task
* @param runnable task
* @throws RejectedExecutionException when The thread pool is full of
*/
public void executeJob(Runnable runnable) throws RejectedExecutionException {
workerExecutor.execute(runnable);
@@ -55,8 +55,6 @@ import static org.dromara.hertzbeat.common.constants.CommonConstants.*;
/**
* Calculate alarms based on the alarm definition rules and collected data
* 根据告警定义规则和采集数据匹配计算告警
*
*
*/
@Component
@@ -96,7 +94,6 @@ public class CalculateAlarm {
this.triggeredAlertMap = new ConcurrentHashMap<>(128);
this.notRecoveredAlertMap = new ConcurrentHashMap<>(128);
// Initialize stateAlertMap
// 初始化stateAlertMap
List<Monitor> monitors = monitorDao.findMonitorsByStatus(CommonConstants.UN_AVAILABLE_CODE);
if (monitors != null) {
for (Monitor monitor : monitors) {
@@ -136,13 +133,12 @@ public class CalculateAlarm {
long monitorId = metricsData.getId();
String app = metricsData.getApp();
String metrics = metricsData.getMetrics();
// If the indicator group whose scheduling priority is 0 has the status of collecting response data UN_REACHABLE/UN_CONNECTABLE, the highest severity alarm is generated to monitor the status change
// 先判断调度优先级为0的指标组采集响应数据状态 UN_REACHABLE/UN_CONNECTABLE 则需发最高级别告警进行任务状态变更
// If the metrics whose scheduling priority is 0 has the status of collecting response data UN_REACHABLE/UN_CONNECTABLE,
// the highest severity alarm is generated to monitor the status change
if (metricsData.getPriority() == 0) {
handlerAvailableMetrics(monitorId, app, metricsData);
}
// Query the alarm definitions associated with the indicator set of the monitoring type
// 查出此监控类型下的此指标集合下关联配置的告警定义信息
// Query the alarm definitions associated with the metrics of the monitoring type
// field - define[]
Map<String, List<AlertDefine>> defineMap = alertDefineService.getMonitorBindAlertDefines(monitorId, app, metrics);
if (defineMap.isEmpty()) {
@@ -293,7 +289,6 @@ public class CalculateAlarm {
.firstAlarmTime(currentTimeMilli)
.lastAlarmTime(currentTimeMilli)
// Keyword matching and substitution in the template
// 模板中关键字匹配替换
.content(AlertTemplateUtil.render(define.getTemplate(), fieldValueMap))
.build();
int defineTimes = define.getTimes() == null ? 1 : define.getTimes();
@@ -22,7 +22,6 @@ import org.springframework.context.annotation.ComponentScan;
/**
*
* @version 2.1
* Created by Musk.Chen on 2023/1/14
*/
@ComponentScan(basePackages = "org.dromara.hertzbeat.alert")
public class AlerterAutoConfiguration {
@@ -34,8 +34,6 @@ import static org.springframework.http.MediaType.APPLICATION_JSON_VALUE;
/**
* Alarm Converge management API
* 告警收敛管理API
*
*
*/
@Tag(name = "Alert Converge API | 告警收敛管理API")
@@ -42,8 +42,6 @@ import static org.springframework.http.MediaType.APPLICATION_JSON_VALUE;
/**
* Converge the batch API for alarms
* 收敛告警批量API
*
*
*/
@Tag(name = "Alert Converge Batch API | 告警收敛管理API")
@@ -45,8 +45,6 @@ import static org.springframework.http.MediaType.APPLICATION_JSON_VALUE;
/**
* Alarm definition management API
* 告警定义管理API
*
*
*/
@Tag(name = "Alert Define API | 告警定义管理API")
@@ -61,7 +59,6 @@ public class AlertDefineController {
@Operation(summary = "New Alarm Definition | 新增告警定义", description = "Added an alarm definition | 新增一个告警定义")
public ResponseEntity<Message<Void>> addNewAlertDefine(@Valid @RequestBody AlertDefine alertDefine) {
// Verify request data
// 校验请求数据
alertDefineService.validate(alertDefine, false);
alertDefineService.addAlertDefine(alertDefine);
return ResponseEntity.ok(Message.success("Add success"));
@@ -71,7 +68,6 @@ public class AlertDefineController {
@Operation(summary = "Modifying an Alarm Definition | 修改告警定义", description = "Modify an existing alarm definition | 修改一个已存在告警定义")
public ResponseEntity<Message<Void>> modifyAlertDefine(@Valid @RequestBody AlertDefine alertDefine) {
// Verify request data
// 校验请求数据
alertDefineService.validate(alertDefine, true);
alertDefineService.modifyAlertDefine(alertDefine);
return ResponseEntity.ok(Message.success("Modify success"));
@@ -83,7 +79,6 @@ public class AlertDefineController {
public ResponseEntity<Message<AlertDefine>> getAlertDefine(
@Parameter(description = "Alarm Definition ID 告警定义ID", example = "6565463543") @PathVariable("id") long id) {
// Obtaining Monitoring Information
// 获取监控信息
AlertDefine alertDefine = alertDefineService.getAlertDefine(id);
if (alertDefine == null) {
return ResponseEntity.ok(Message.fail(MONITOR_NOT_EXIST_CODE, "AlertDefine not exist."));
@@ -98,7 +93,6 @@ public class AlertDefineController {
public ResponseEntity<Message<Void>> deleteAlertDefine(
@Parameter(description = "Alarm Definition ID 告警定义ID", example = "6565463543") @PathVariable("id") long id) {
// If the alarm definition does not exist or is deleted successfully, the deletion succeeds
// 删除告警定义不存在或删除成功都返回成功
alertDefineService.deleteAlertDefine(id);
return ResponseEntity.ok(Message.success("Delete success"));
}
@@ -46,8 +46,6 @@ import static org.springframework.http.MediaType.APPLICATION_JSON_VALUE;
/**
* Define the batch API for alarms
* 告警定义批量API
*
*
*/
@Tag(name = "Alert Define Batch API | 告警定义管理API")
@@ -111,7 +109,6 @@ public class AlertDefinesController {
Predicate[] predicates = new Predicate[andList.size()];
return criteriaBuilder.and(andList.toArray(predicates));
};
// 分页是必须的
Sort sortExp = Sort.by(new Sort.Order(Sort.Direction.fromString(order), sort));
PageRequest pageRequest = PageRequest.of(pageIndex, pageSize, sortExp);
Page<AlertDefine> alertDefinePage = alertDefineService.getAlertDefines(specification, pageRequest);
@@ -20,8 +20,8 @@ import java.util.Optional;
import static org.springframework.http.MediaType.APPLICATION_JSON_VALUE;
/**
* Extern Alarm Manage API
*
* 第三方告警上报接口
*/
@Tag(name = "Extern Alarm Manage API | 第三方告警管理API")
@RestController
@@ -33,8 +33,7 @@ public class AlertReportController {
private AlertService alertService;
@PostMapping("/{cloud}")
@Operation(summary = "Interface for reporting external alarm information of cloud service 对外上报告警信息 接口",
description = "对外 新增一个云服务告警")
@Operation(summary = "Interface for reporting external alarm information of cloud service 对外上报告警信息接口")
public ResponseEntity<Message<Void>> addNewAlertReportFromCloud(@PathVariable("cloud") String cloudServiceName,
@RequestBody String alertReport) {
// 根据枚举获取到对应的枚举对象
@@ -35,8 +35,6 @@ import static org.springframework.http.MediaType.APPLICATION_JSON_VALUE;
/**
* Alarm Silence management API
* 告警静默管理API
*
*
*/
@Tag(name = "Alert Silence API | 告警静默管理API")
@@ -42,8 +42,6 @@ import static org.springframework.http.MediaType.APPLICATION_JSON_VALUE;
/**
* Silence the batch API for alarms
* 静默告警批量API
*
*
*/
@Tag(name = "Alert Silence Batch API | 告警静默管理API")
@@ -41,9 +41,7 @@ import java.util.List;
import static org.springframework.http.MediaType.APPLICATION_JSON_VALUE;
/**
* Alarm Management API 告警管理API
*
*
* Alarm Management API
*
*
*/
@@ -25,8 +25,7 @@ import org.springframework.data.jpa.repository.Modifying;
import java.util.Set;
/**
* AlertConverge 数据库操作
*
* AlertConverge Dao
*
*/
public interface AlertConvergeDao extends JpaRepository<AlertConverge, Long>, JpaSpecificationExecutor<AlertConverge> {
@@ -29,26 +29,21 @@ import java.util.List;
import java.util.Set;
/**
* Alert Database Operations Alert数据库表操作
*
*
* Alert Database Operations
*
*/
public interface AlertDao extends JpaRepository<Alert, Long>, JpaSpecificationExecutor<Alert> {
/**
* Delete alerts based on ID list 根据ID列表删除告警
*
* @param alertIds Alert ID List 告警ID列表
* Delete alerts based on ID list
* @param alertIds Alert ID List
*/
void deleteAlertsByIdIn(Set<Long> alertIds);
/**
* Updates the alarm status based on the alarm ID-status value
* 根据告警ID-状态值 更新告警状态
*
* @param status 状态值
* @param ids 告警ID列表
* @param status status value
* @param ids alarm ids
*/
@Modifying
@Query("update Alert set status = :status where id in :ids")
@@ -57,8 +52,7 @@ public interface AlertDao extends JpaRepository<Alert, Long>, JpaSpecificationEx
/**
* Query the number of unhandled alarms of each alarm severity
* 查询各个告警级别的未处理告警数量
*
* @return List of alerts num 告警数量
* @return List of alerts num
*/
@Query("select new org.dromara.hertzbeat.alert.dto.AlertPriorityNum(mo.priority, count(mo.id)) from Alert mo where mo.status = 0 group by mo.priority")
List<AlertPriorityNum> findAlertPriorityNum();
@@ -25,43 +25,33 @@ import java.util.List;
import java.util.Set;
/**
* AlertDefineBind database operations 数据库操作
*
*
* AlertDefineBind database operations
*
*/
public interface AlertDefineBindDao extends JpaRepository<AlertDefineMonitorBind, Long>, JpaSpecificationExecutor<AlertDefineMonitorBind> {
/**
* Delete the alarm definition and monitor association based on the alarm definition ID
* 根据告警定义ID删除告警定义与监控关联
*
* @param alertDefineId Alarm Definition ID 告警定义ID
* @param alertDefineId Alarm Definition ID
*/
void deleteAlertDefineBindsByAlertDefineIdEquals(Long alertDefineId);
/**
* Deleting alarms based on monitoring IDs defines monitoring associations
* 根据监控任务ID删除告警定义监控关联
*
* @param monitorId Monitor Id 监控任务ID
* @param monitorId Monitor Id
*/
void deleteAlertDefineMonitorBindsByMonitorIdEquals(Long monitorId);
/**
* Delete alarm definition monitoring association based on monitoring ID list
* 根据监控任务ID列表删除告警定义监控关联
*
* @param monitorIds Monitoring ID List 监控任务ID列表
* @param monitorIds Monitoring ID List
*/
void deleteAlertDefineMonitorBindsByMonitorIdIn(Set<Long> monitorIds);
/**
* Query monitoring related information based on alarm definition ID
* 根据告警定义ID查询监控关联信息
*
* @param alertDefineId Alarm Definition ID 告警定义ID
* @return Associated monitoring information 关联监控信息
* @param alertDefineId Alarm Definition ID
* @return Associated monitoring information
*/
List<AlertDefineMonitorBind> getAlertDefineBindsByAlertDefineIdEquals(Long alertDefineId);
}
@@ -27,25 +27,23 @@ import java.util.List;
import java.util.Set;
/**
* AlertDefine 数据库操作
*
* AlertDefine Dao
*
*/
public interface AlertDefineDao extends JpaRepository<AlertDefine, Long>, JpaSpecificationExecutor<AlertDefine> {
/**
* Delete alarm definitions based on the ID list
* 根据ID列表删除告警定义
* @param alertDefineIds 告警定义ID列表
* @param alertDefineIds alarm define ids
*/
void deleteAlertDefinesByIdIn(Set<Long> alertDefineIds);
/**
* Query the default alarm thresholds based on the monitoring indicator type
* Query the default alarm thresholds based on the monitoring metrics type
* 根据监控指标类型查询对应默认告警定义阈值
* @param app 监控类型
* @param metric 指标集合类型
* @return The alarm is defined 告警定义
* @param app monitoring type
* @param metric metrics
* @return alarm defines
*/
List<AlertDefine> queryAlertDefinesByAppAndMetricAndPresetTrueAndEnableTrue(String app, String metric);
@@ -61,10 +59,10 @@ public interface AlertDefineDao extends JpaRepository<AlertDefine, Long>, JpaSpe
/**
* Query the alarm definition list associated with the monitoring ID
* 根据监控任务ID查询与之关联的告警定义列表
* @param monitorId 监控任务ID
* @param app 监控类型
* @param metrics 指标组
* @return Alarm Definition List 告警定义列表
* @param monitorId monitor id
* @param app monitor type
* @param metrics metrics
* @return Alarm Definition List
*/
@Query("select define from AlertDefine define join AlertDefineMonitorBind bind on bind.alertDefineId = define.id " +
"where bind.monitorId = :monitorId and define.app = :app and define.metric = :metrics and define.enable = true and define.preset = false")
@@ -27,24 +27,15 @@ import org.springframework.data.repository.query.Param;
import java.util.List;
/**
* Alert Monitor 数据库操作
*
* Alert Monitor Dao
*
*/
public interface AlertMonitorDao extends JpaRepository<Monitor, Long>, JpaSpecificationExecutor<Monitor> {
/**
* Query the monitoring status of a specified monitoring state 查询指定任务状态的监控
* @param status 任务状态
* @return Monitor the list 监控列表
*/
List<Monitor> findMonitorsByStatusIn(List<Byte> status);
/**
* Query the monitoring status of a specified monitoring state 查询指定任务状态的监控
* @param status 任务状态
* @return Monitor the list 监控列表
* Query the monitoring status of a specified monitoring state
* @param status status value
* @return Monitor the list
*/
List<Monitor> findMonitorsByStatus(Byte status);
@@ -25,8 +25,7 @@ import org.springframework.data.jpa.repository.Modifying;
import java.util.Set;
/**
* AlertSilence 数据库操作
*
* AlertSilence Dao
*
*/
public interface AlertSilenceDao extends JpaRepository<AlertSilence, Long>, JpaSpecificationExecutor<AlertSilence> {
@@ -21,9 +21,7 @@ import lombok.AllArgsConstructor;
import lombok.Data;
/**
* Number of monitoring level alarms 监控级别告警数量
*
*
* Number of monitoring level alarms
*
*/
@Data
@@ -31,12 +29,12 @@ import lombok.Data;
public class AlertPriorityNum {
/**
* Alarm level 告警级别
* Alarm level
*/
private byte priority;
/**
* count 数量
* Alarm count
*/
private long num;
}
@@ -26,9 +26,7 @@ import static io.swagger.v3.oas.annotations.media.Schema.AccessMode.READ_ONLY;
/**
* Alarm Statistics Information 告警统计信息
*
*
* Alarm Statistics Information
*
*/
@Data
@@ -8,7 +8,7 @@ import org.dromara.hertzbeat.alert.dto.TenCloudAlertReport;
import java.util.Arrays;
/**
* 云服务告警枚举
* Cloud server alarm enum
*/
@AllArgsConstructor
@Getter
@@ -17,12 +17,12 @@ public enum CloudServiceAlarmInformationEnum {
TencentCloud("tencloud", TenCloudAlertReport.class);
/**
* 云服务名称
* cloud service name
*/
private final String cloudServiceName;
/**
* 云服务对应的请求实体
* cloud service body
*/
private final Class<? extends CloudAlertReportAbstract> cloudServiceAlarmInformationEntity;
@@ -16,7 +16,6 @@ import java.util.Map;
/**
* reduce alarm and send alert data
*
*
*/
@Service
@RequiredArgsConstructor
@@ -16,24 +16,25 @@ import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
/**
* alarm converge
* 告警收敛
* alarm converge
*
*
*/
@Service
public class AlarmConvergeReduce {
private final AlertConvergeDao alertConvergeDao;
private final Map<Integer, Alert> converageAlertMap;
public AlarmConvergeReduce(AlertConvergeDao alertConvergeDao) {
this.alertConvergeDao = alertConvergeDao;
this.converageAlertMap = new ConcurrentHashMap<>(16);
}
/**
* currentAlert converge filter data
*
* @param currentAlert currentAlert
* @return true when not filter
*/
@@ -109,9 +110,9 @@ public class AlarmConvergeReduce {
if (evalInterval <= 0) {
return true;
}
int alertHash = Objects.hash(currentAlert.getPriority())
+ Arrays.hashCode(currentAlert.getTags().keySet().toArray(new String[0]))
+ Arrays.hashCode(currentAlert.getTags().values().toArray(new String[0]));
int alertHash = Objects.hash(currentAlert.getPriority())
+ Arrays.hashCode(currentAlert.getTags().keySet().toArray(new String[0]))
+ Arrays.hashCode(currentAlert.getTags().values().toArray(new String[0]));
Alert preAlert = converageAlertMap.get(alertHash);
if (preAlert == null) {
currentAlert.setTimes(1);
@@ -19,7 +19,6 @@ import java.util.Optional;
/**
* silence alarm
*
*
*/
@Service
@RequiredArgsConstructor
@@ -10,29 +10,27 @@ import java.util.Set;
/**
* management interface service for alert converge
*
*
*/
public interface AlertConvergeService {
/**
* Verify the correctness of the request data parameters
* 校验请求数据参数正确性
* @param alertConverge AlertConverge
* @param isModify 是否是修改配置
* @throws IllegalArgumentException A checksum parameter error is thrown 校验参数错误抛出
* @param isModify whether modify
* @throws IllegalArgumentException A checksum parameter error is thrown
*/
void validate(AlertConverge alertConverge, boolean isModify) throws IllegalArgumentException;
/**
* New AlertConverge
* @param alertConverge AlertConverge Entity 收敛策略实体
* @throws RuntimeException Added procedure exception throwing 新增过程异常抛出
* @param alertConverge AlertConverge Entity
* @throws RuntimeException Added procedure exception throwing
*/
void addAlertConverge(AlertConverge alertConverge) throws RuntimeException;
/**
* Modifying an AlertConverge 修改收敛策略
* @param alertConverge Alarm definition Entity 收敛策略实体
* @throws RuntimeException Exception thrown during modification 修改过程中异常抛出
* Modifying an AlertConverge
* @param alertConverge Alarm definition Entity
* @throws RuntimeException Exception thrown during modification
*/
void modifyAlertConverge(AlertConverge alertConverge) throws RuntimeException;
@@ -40,24 +38,23 @@ public interface AlertConvergeService {
* Obtain AlertConverge information
* @param convergeId AlertConverge ID
* @return AlertConverge
* @throws RuntimeException An exception was thrown during the query 查询过程中异常抛出
* @throws RuntimeException An exception was thrown during the query
*/
AlertConverge getAlertConverge(long convergeId) throws RuntimeException;
/**
* Delete AlertConverge in batches 批量删除收敛策略
* @param convergeIds AlertConverge IDs 收敛策略IDs
* @throws RuntimeException Exception thrown during deletion 删除过程中异常抛出
* Delete AlertConverge in batches
* @param convergeIds AlertConverge IDs
* @throws RuntimeException Exception thrown during deletion
*/
void deleteAlertConverges(Set<Long> convergeIds) throws RuntimeException;
/**
* Dynamic conditional query
* 动态条件查询
* @param specification Query conditions 查询条件
* @param pageRequest Paging parameters 分页参数
* @return The query results 查询结果
* @param specification Query conditions
* @param pageRequest Paging parameters
* @return The query results
*/
Page<AlertConverge> getAlertConverges(Specification<AlertConverge> specification, PageRequest pageRequest);
}
@@ -28,110 +28,104 @@ import java.util.Map;
import java.util.Set;
/**
* 告警定义管理接口
*
* Alarm define manager service
*
*/
public interface AlertDefineService {
/**
* Verify the correctness of the request data parameters
* 校验请求数据参数正确性
* @param alertDefine alertDefine
* @param isModify 是否是修改配置
* @throws IllegalArgumentException A checksum parameter error is thrown 校验参数错误抛出
* @param isModify whether modify
* @throws IllegalArgumentException A checksum parameter error is thrown
*/
void validate(AlertDefine alertDefine, boolean isModify) throws IllegalArgumentException;
/**
* New Alarm Definition
* 新增告警定义
* @param alertDefine Alarm definition Entity 告警定义实体
* @throws RuntimeException Added procedure exception throwing 新增过程异常抛出
* @param alertDefine Alarm definition Entity
* @throws RuntimeException Added procedure exception throwing
*/
void addAlertDefine(AlertDefine alertDefine) throws RuntimeException;
/**
* Modifying an Alarm Definition 修改告警定义
* @param alertDefine Alarm definition Entity 告警定义实体
* @throws RuntimeException Exception thrown during modification 修改过程中异常抛出
* Modifying an Alarm Definition
* @param alertDefine Alarm definition Entity
* @throws RuntimeException Exception thrown during modification
*/
void modifyAlertDefine(AlertDefine alertDefine) throws RuntimeException;
/**
* Deleting an Alarm Definition
* 删除告警定义
* @param alertId Alarm Definition ID 告警定义ID
* @throws RuntimeException Exception thrown during deletion 删除过程中异常抛出
* @param alertId Alarm Definition ID
* @throws RuntimeException Exception thrown during deletion
*/
void deleteAlertDefine(long alertId) throws RuntimeException;
/**
* Obtain alarm definition information
* 获取告警定义信息
* @param alertId Monitor the ID 监控任务ID
* @param alertId Monitor the ID
* @return AlertDefine
* @throws RuntimeException An exception was thrown during the query 查询过程中异常抛出
* @throws RuntimeException An exception was thrown during the query
*/
AlertDefine getAlertDefine(long alertId) throws RuntimeException;
/**
* Delete alarm definitions in batches 批量删除告警定义
* @param alertIds Alarm Definition IDs 告警定义IDs
* @throws RuntimeException Exception thrown during deletion 删除过程中异常抛出
* Delete alarm definitions in batches
* @param alertIds Alarm Definition IDs
* @throws RuntimeException Exception thrown during deletion
*/
void deleteAlertDefines(Set<Long> alertIds) throws RuntimeException;
/**
* Dynamic conditional query 动态条件查询
* @param specification Query conditions 查询条件
* @param pageRequest Paging parameters 分页参数
* @return The query results 查询结果
* Dynamic conditional query
* @param specification Query conditions
* @param pageRequest Paging parameters
* @return The query results
*/
Page<AlertDefine> getMonitorBindAlertDefines(Specification<AlertDefine> specification, PageRequest pageRequest);
/**
* Association between application alarm schedule and monitoring |应用告警定于与监控关联关系
* @param alertId Alarm Definition ID 告警定义ID
* @param alertDefineBinds correlation 关联关系
* Association between application alarm schedule and monitoring
* @param alertId Alarm Definition ID
* @param alertDefineBinds correlation
*/
void applyBindAlertDefineMonitors(Long alertId, List<AlertDefineMonitorBind> alertDefineBinds);
/**
* Query the alarm definitions that match the specified indicator group associated with the monitoring ID
* 查询与此监控任务ID关联的指定指标匹配的告警定义
* @param monitorId Monitor the ID 监控任务ID
* @param app Monitoring type 监控类型
* @param metrics Index group 指标组
* Query the alarm definitions that match the specified metrics associated with the monitoring ID
* 查询与此监控任务ID关联的指定指标匹配的告警定义
* @param monitorId Monitor the ID
* @param app Monitoring type
* @param metrics metrics
* @return field - define[]
*/
Map<String, List<AlertDefine>> getMonitorBindAlertDefines(long monitorId, String app, String metrics);
/**
* Query the alarm definitions that match the specified indicator group associated with the monitoring ID
* Query the alarm definitions that match the specified metrics associated with the monitoring ID
* 查询与此监控任务ID关联的可用性告警定义
* @param monitorId Monitor the ID 监控任务ID
* @param app Monitoring type 监控类型
* @param metrics Index group 指标组
* @param monitorId Monitor the ID
* @param app Monitoring type
* @param metrics metrics
* @return field - define[]
*/
AlertDefine getMonitorBindAlertAvaDefine(long monitorId, String app, String metrics);
/**
* Dynamic conditional query
* 动态条件查询
* @param specification Query conditions 查询条件
* @param pageRequest Paging parameters 分页参数
* @return The query results 查询结果
* @param specification Query conditions
* @param pageRequest Paging parameters
* @return The query results
*/
Page<AlertDefine> getAlertDefines(Specification<AlertDefine> specification, PageRequest pageRequest);
/**
* Query the associated monitoring list information based on the alarm definition ID
* 根据告警定义ID查询其关联的监控列表关联信息
* @param alertDefineId Alarm Definition ID 告警定义ID
* @return Associated information about the monitoring list 监控列表关联信息
* @param alertDefineId Alarm Definition ID
* @return Associated information about the monitoring list
*/
List<AlertDefineMonitorBind> getBindAlertDefineMonitors(long alertDefineId);
}
@@ -29,76 +29,61 @@ import java.util.List;
/**
* Alarm information management interface
* 告警信息管理接口
*
*
*
*/
public interface AlertService {
/**
* Add alarm record
* 新增告警记录
*
* @param alert Alert entity 告警实体
* @throws RuntimeException Add process exception throw 新增过程异常抛出
* @param alert Alert entity
* @throws RuntimeException Add process exception throw
*/
void addAlert(Alert alert) throws RuntimeException;
/**
* Dynamic conditional query
* 动态条件查询
*
* @param specification Query conditions 查询条件
* @param pageRequest pagination parameters 分页参数
* @return search result 查询结果
* @param specification Query conditions
* @param pageRequest pagination parameters
* @return search result
*/
Page<Alert> getAlerts(Specification<Alert> specification, PageRequest pageRequest);
/**
* Delete alarms in batches according to the alarm ID list
* 根据告警ID列表批量删除告警
*
* @param ids Alarm ID List 告警IDS
* @param ids Alarm ID List
*/
void deleteAlerts(HashSet<Long> ids);
/**
* Clear all alerts
* 清空所有告警记录
*/
void clearAlerts();
/**
* Update the alarm status according to the alarm ID-status value
* 根据告警ID-状态值 更新告警状态
*
* @param status Alarm status to be modified 待修改为的告警状态
* @param ids Alarm ID List to be modified 待修改的告警ID集合
* @param status Alarm status to be modified
* @param ids Alarm ID List to be modified
*/
void editAlertStatus(Byte status, List<Long> ids);
/**
* Get alarm statistics information 获取告警统计信息
*
* @return Alarm statistics information 告警统计
* Get alarm statistics information
* @return Alarm statistics information
*/
AlertSummary getAlertsSummary();
/**
* A third party reports an alarm 第三方 上报告警信息
* @param alertReport The alarm information 告警信息
* A third party reports an alarm
* @param alertReport The alarm information
*/
void addNewAlertReport(AlertReport alertReport);
/**
* Dynamic conditional query
* 动态条件查询
*
* @param specification Query conditions 查询条件
* @return search result 查询结果
* @param specification Query conditions
* @return search result
*/
List<Alert> getAlerts(Specification<Alert> specification);
}
@@ -10,29 +10,27 @@ import java.util.Set;
/**
* management interface service for alert silence
*
*
*/
public interface AlertSilenceService {
/**
* Verify the correctness of the request data parameters
* 校验请求数据参数正确性
* @param alertSilence AlertSilence
* @param isModify 是否是修改配置
* @throws IllegalArgumentException A checksum parameter error is thrown 校验参数错误抛出
* @param isModify whether modify
* @throws IllegalArgumentException A checksum parameter error is thrown
*/
void validate(AlertSilence alertSilence, boolean isModify) throws IllegalArgumentException;
/**
* New AlertSilence
* @param alertSilence AlertSilence Entity 静默策略实体
* @throws RuntimeException Added procedure exception throwing 新增过程异常抛出
* @param alertSilence AlertSilence Entity
* @throws RuntimeException Added procedure exception throwing
*/
void addAlertSilence(AlertSilence alertSilence) throws RuntimeException;
/**
* Modifying an AlertSilence 修改静默策略
* @param alertSilence Alarm definition Entity 静默策略实体
* @throws RuntimeException Exception thrown during modification 修改过程中异常抛出
* Modifying an AlertSilence
* @param alertSilence Alarm definition Entity
* @throws RuntimeException Exception thrown during modification
*/
void modifyAlertSilence(AlertSilence alertSilence) throws RuntimeException;
@@ -40,24 +38,23 @@ public interface AlertSilenceService {
* Obtain AlertSilence information
* @param silenceId AlertSilence ID
* @return AlertSilence
* @throws RuntimeException An exception was thrown during the query 查询过程中异常抛出
* @throws RuntimeException An exception was thrown during the query
*/
AlertSilence getAlertSilence(long silenceId) throws RuntimeException;
/**
* Delete AlertSilence in batches 批量删除静默策略
* @param silenceIds AlertSilence IDs 静默策略IDs
* @throws RuntimeException Exception thrown during deletion 删除过程中异常抛出
* Delete AlertSilence in batches
* @param silenceIds AlertSilence IDs
* @throws RuntimeException Exception thrown during deletion
*/
void deleteAlertSilences(Set<Long> silenceIds) throws RuntimeException;
/**
* Dynamic conditional query
* 动态条件查询
* @param specification Query conditions 查询条件
* @param pageRequest Paging parameters 分页参数
* @return The query results 查询结果
* @param specification Query conditions
* @param pageRequest Paging parameters
* @return The query results
*/
Page<AlertSilence> getAlertSilences(Specification<AlertSilence> specification, PageRequest pageRequest);
}
@@ -19,7 +19,6 @@ import java.util.Set;
/**
* implement for alert converge service
*
*
*/
@Service
@Transactional(rollbackFor = Exception.class)
@@ -37,8 +37,6 @@ import java.util.stream.Collectors;
/**
* Alarm definition management interface implementation
* 告警定义管理接口实现
*
*
*/
@Service
@@ -101,10 +99,8 @@ public class AlertDefineServiceImpl implements AlertDefineService {
// todo 校验此告警定义和监控是否存在
// Delete all associations of this alarm
// 先删除此告警的所有关联
alertDefineBindDao.deleteAlertDefineBindsByAlertDefineIdEquals(alertId);
// Save the associated
// 保存关联
alertDefineBindDao.saveAll(alertDefineBinds);
}
@@ -43,9 +43,7 @@ import java.util.List;
import java.util.Map;
/**
* Realization of Alarm Information Service 告警信息服务实现
*
*
* Realization of Alarm Information Service
*
*/
@Service
@@ -87,8 +85,7 @@ public class AlertServiceImpl implements AlertService {
@Override
public AlertSummary getAlertsSummary() {
AlertSummary alertSummary = new AlertSummary();
//Statistics on the alarm information in the alarm state
//统计正在告警状态下的告警信息
// Statistics on the alarm information in the alarm state
List<AlertPriorityNum> priorityNums = alertDao.findAlertPriorityNum();
if (priorityNums != null) {
for (AlertPriorityNum priorityNum : priorityNums) {
@@ -140,9 +137,9 @@ public class AlertServiceImpl implements AlertService {
}
/**
* The external alarm information is converted to Alert 对外告警信息 转换为Alert
* @param alertReport 对外告警信息
* @return Alert entity Alert实体
* The external alarm information is converted to Alert
* @param alertReport alarm body
* @return Alert entity
*/
private Alert buildAlertData(AlertReport alertReport){
Map<String, String> annotations = alertReport.getAnnotations();
@@ -20,7 +20,6 @@ import java.util.Set;
/**
* management interface service implement for alert silence
*
*
*/
@Service
@Transactional(rollbackFor = Exception.class)
@@ -27,7 +27,6 @@ import java.util.regex.Pattern;
* Alarm template keyword matching replacement engine tool
* 告警模版关键字匹配替换引擎工具
*
*
*/
@Slf4j
public class AlertTemplateUtil {
@@ -1,11 +1,14 @@
package org.dromara.hertzbeat.alert.util;
import lombok.extern.slf4j.Slf4j;
import java.text.ParseException;
import java.text.SimpleDateFormat;
/**
* 记录一些常用的日期格式
* date time common util
*/
@Slf4j
public class DateUtil {
private static final String[] DATE_FORMATS = {
@@ -13,7 +16,8 @@ public class DateUtil {
"yyyy-MM-dd HH:mm:ss"};
/**
* 常用日期格式转时间戳
* convert date to timestamp
* @param date date
*/
public static Long getTimeStampFromSomeFormats(String date) {
SimpleDateFormat sdf;
@@ -21,21 +25,24 @@ public class DateUtil {
try {
sdf = new SimpleDateFormat(dateFormat);
return sdf.parse(date).getTime();
} catch (ParseException e) {}
} catch (ParseException e) {
log.error(e.getMessage());
}
}
return null;
}
/**
* 指定日期格式转换时间戳
* convert format data to timestamp
*/
public static Long getTimeStampFromFormat(String date, String format) {
SimpleDateFormat sdf = new SimpleDateFormat(format);
try {
return sdf.parse(date).getTime();
} catch (Exception e) {
throw new RuntimeException("时间格式解析异常!");
log.error(e.getMessage());
}
return null;
}
}
@@ -22,23 +22,19 @@ import org.dromara.hertzbeat.common.entity.job.Metrics;
import org.dromara.hertzbeat.common.entity.message.CollectRep;
/**
* Specific metrics group collection implementation abstract class
*
*
* Specific metrics collection implementation abstract class
*
*/
public abstract class AbstractCollect {
/**
* Real acquisition implementation interface
*
* @param builder response builder
* @param appId App monitoring ID
* @param app Application Type
* @param metrics Metric group configuration
* return response builder
* @param monitorId monitor id
* @param app monitor type
* @param metrics metric configuration
*/
public abstract void collect(CollectRep.MetricsData.Builder builder, long appId, String app, Metrics metrics);
public abstract void collect(CollectRep.MetricsData.Builder builder, long monitorId, String app, Metrics metrics);
/**
* the protocol this collect instance support
@@ -18,14 +18,13 @@
package org.dromara.hertzbeat.collector.collect.common.cache;
/**
* 连接资源关闭回调接口
*
* resource in cache remove callback
*
*/
public interface CacheCloseable {
/**
* 在缓存remove掉此对象前,回调接口对连接对象进行相关资源的释放
* when the resource in cache want to be removed, callback this
*/
void close();
}
@@ -23,8 +23,7 @@ import lombok.Data;
import java.util.Objects;
/**
* 缓存key唯一标识符
*
* resource identifier in cache
*
*/
@Data
@@ -30,50 +30,45 @@ import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
* lru cache 对连接对象进行缓存
*
* lru common resource cache
*
*/
@Slf4j
public class CommonCache {
/**
* 默认缓存时间 800s
* default cache time 800s
*/
private static final long DEFAULT_CACHE_TIMEOUT = 800 * 1000L;
/**
* 默认最大缓存数量
* default cache num
*/
private static final int DEFAULT_MAX_CAPACITY = 10000;
/**
* cacheTime数组大小
* cacheTime length
*/
private static final int CACHE_TIME_LENGTH = 2;
/**
* 存储对象的数据过期时间点
* cache timeout map
*/
private Map<Object, Long[]> timeoutMap;
/**
* 存储缓存对象
* object cache
*/
private ConcurrentLinkedHashMap<Object, Object> cacheMap;
/**
* 过期数据清理线程池
* the executor who clean cache when timeout
*/
private ThreadPoolExecutor cleanTimeoutExecutor;
private CommonCache() { init();}
/**
* 初始化 cache
*/
private void init() {
// 初始化lru hashmap
cacheMap = new ConcurrentLinkedHashMap
.Builder<>()
.maximumWeightedCapacity(DEFAULT_MAX_CAPACITY)
@@ -84,17 +79,12 @@ public class CommonCache {
}
log.info("lru cache discard key: {}, value: {}.", key, value);
}).build();
// 初始化时间纪录map
timeoutMap = new ConcurrentHashMap<>(DEFAULT_MAX_CAPACITY >> 6);
// 初始化过期数据清理线程池
cleanTimeoutExecutor = new ThreadPoolExecutor(1, 1,
1, TimeUnit.SECONDS,
new ArrayBlockingQueue<>(1), r -> new Thread("lru-cache-timeout-cleaner"),
new ThreadPoolExecutor.DiscardOldestPolicy());
// 初始化可用性探测定位任务,每次探测间隔时间为20分钟
// init monitor available detector cyc task
ScheduledThreadPoolExecutor scheduledExecutor = new ScheduledThreadPoolExecutor(1,
r -> new Thread(r, "lru-cache-available-detector"));
scheduledExecutor.scheduleWithFixedDelay(this::detectCacheAvailable,
@@ -102,12 +92,11 @@ public class CommonCache {
}
/**
* 探测所有可探测的缓存对象的可用性,清除不可用和过期对象
* detect all cache available, cleanup not ava object
*/
private void detectCacheAvailable() {
try {
cacheMap.forEach((key, value) -> {
// 先判断是否过期
Long[] cacheTime = timeoutMap.get(key);
long currentTime = System.currentTimeMillis();
if (cacheTime == null || cacheTime.length != CACHE_TIME_LENGTH
@@ -126,7 +115,7 @@ public class CommonCache {
}
/**
* 清理过期线程
* clean timeout cache
*/
private void cleanTimeoutCache() {
try {
@@ -137,7 +126,7 @@ public class CommonCache {
if (cacheTime == null || cacheTime.length != CACHE_TIME_LENGTH) {
timeoutMap.put(key, new Long[]{currentTime, DEFAULT_CACHE_TIMEOUT});
} else if (cacheTime[0] + cacheTime[1] < currentTime) {
// 过期了 discard 关闭这个cache的资源
// timeout, remove this object cache
log.warn("[cache] clean the timeout cache, key {}", key);
timeoutMap.remove(key);
cacheMap.remove(key);
@@ -153,10 +142,10 @@ public class CommonCache {
}
/**
* 新增或更新cache
* @param key 存储对象key
* @param value 存储对象
* @param timeDiff 缓存对象保存时间 millis
* add update cache
* @param key cache key
* @param value cache value
* @param timeDiff cache time millis
*/
public void addCache(Object key, Object value, Long timeDiff) {
removeCache(key);
@@ -176,19 +165,19 @@ public class CommonCache {
}
/**
* 新增或更新cache
* @param key 存储对象key
* @param value 存储对象
* add update cache
* @param key cache key
* @param value cache value
*/
public void addCache(Object key, Object value) {
addCache(key, value, DEFAULT_CACHE_TIMEOUT);
}
/**
* 根据缓存key获取缓存对象
* @param key key
* @param refreshCache 是否刷新命中的缓存的存活时间 true是,false否
* @return 缓存对象
* get cache by key
* @param key cache key
* @param refreshCache is refresh cache
* @return cache object
*/
public Optional<Object> getCache(Object key, boolean refreshCache) {
Long[] cacheTime = timeoutMap.get(key);
@@ -215,7 +204,7 @@ public class CommonCache {
}
/**
* 根据缓存key删除缓存对象
* remove cache by key
* @param key key
*/
public void removeCache(Object key) {
@@ -227,7 +216,7 @@ public class CommonCache {
}
/**
* 获取缓存实例
* get common cache instance
* @return cache
*/
public static CommonCache getInstance() {
@@ -235,12 +224,9 @@ public class CommonCache {
}
/**
* 静态内部类
* static instance
*/
private static class SingleInstance {
/**
* 单例
*/
private static final CommonCache INSTANCE= new CommonCache();
}
}
@@ -24,7 +24,6 @@ import java.sql.Connection;
/**
* jdbc common connection
*
*
*/
@Slf4j
public class JdbcConnect implements CacheCloseable {
@@ -5,9 +5,7 @@ import lombok.extern.slf4j.Slf4j;
import javax.management.remote.JMXConnector;
/**
* jmx链接销毁管理
*
*
* jmx connect object
*
**/
@Slf4j
@@ -6,8 +6,6 @@ import lombok.extern.slf4j.Slf4j;
/**
* mongodb connect client
*
*
*
*/
@Slf4j
public class MongodbConnect implements CacheCloseable {
@@ -23,8 +23,6 @@ import lombok.extern.slf4j.Slf4j;
/**
* redis connection
*
*
*
*/
@Slf4j
public class RedisConnect implements CacheCloseable {
@@ -6,8 +6,6 @@ import org.apache.sshd.client.session.ClientSession;
/**
* ssh connection holder
*
*
*
*/
@Slf4j
public class SshConnect implements CacheCloseable {
@@ -40,8 +40,7 @@ import java.util.Date;
import java.util.concurrent.TimeUnit;
/**
* 统一的http客户端连接池
*
* common http client
*
*/
@Slf4j
@@ -52,50 +51,51 @@ public class CommonHttpClient {
private static PoolingHttpClientConnectionManager connectionManager;
/**
* 此连接池所能提供的最大连接数
* all max total connection
*/
private static final int MAX_TOTAL_CONNECTIONS = 50000;
/**
* 每个路由所能分配的最大连接数
* peer route max total connection
*/
private static final int MAX_PER_ROUTE_CONNECTIONS = 80;
/**
* 从连接池中获取连接的默认超时时间 4秒
* timeout for get connect from pool(ms)
*/
private static final int REQUIRE_CONNECT_TIMEOUT = 4000;
/**
* 双端建立连接超时时间 4秒
* tcp connect timeout(ms)
*/
private static final int CONNECT_TIMEOUT = 4000;
/**
* socketReadTimeout 响应tcp报文的最大间隔超时时间
* socket read timeout(ms)
*/
private static final int SOCKET_TIMEOUT = 60000;
/**
* validated time for idle connection
* 空闲连接免检的有效时间,被重用的空闲连接若超过此时间,需检查此连接的可用性
*/
private static final int INACTIVITY_VALIDATED_TIME = 10000;
/**
* ssl版本
* ssl supported version
*/
private static final String[] SUPPORTED_SSL = {"TLSv1","TLSv1.1","TLSv1.2","SSLv3"};
static {
try {
// 初始化ssl上下文
SSLContext sslContext = SSLContexts.createDefault();
X509TrustManager x509TrustManager = new X509TrustManager() {
@Override
public void checkClientTrusted(X509Certificate[] x509Certificates, String s) throws CertificateException { }
@Override
public void checkServerTrusted(X509Certificate[] x509Certificates, String s) throws CertificateException {
// 判断服务器证书有效期时间
// check server certificate timeout
// 判断服务器证书有效时间
Date now = new Date();
if (x509Certificates != null && x509Certificates.length > 0) {
for (X509Certificate certificate : x509Certificates) {
@@ -110,30 +110,23 @@ public class CommonHttpClient {
public X509Certificate[] getAcceptedIssuers() { return null; }
};
sslContext.init(null, new TrustManager[]{x509TrustManager}, null);
// 设置支持的ssl版本
SSLConnectionSocketFactory sslFactory = new SSLConnectionSocketFactory(sslContext, SUPPORTED_SSL, null, new NoopHostnameVerifier());
// 注册 http https
Registry<ConnectionSocketFactory> registry = RegistryBuilder.<ConnectionSocketFactory>create()
.register("http", PlainConnectionSocketFactory.INSTANCE)
.register("https", sslFactory)
.build();
// 网络请求默认配置
RequestConfig requestConfig = RequestConfig.custom()
// 从连接池获取连接超时时间
.setConnectionRequestTimeout(REQUIRE_CONNECT_TIMEOUT)
// 和对端新连接建立时间,三次握手时间
.setConnectTimeout(CONNECT_TIMEOUT)
// 数据传输最大响应间隔时间
.setSocketTimeout(SOCKET_TIMEOUT)
// 遇到301 302自动重定向跳转
// auto redirect when 301 302 response status
.setRedirectsEnabled(true)
.build();
// 连接池
// connection pool
connectionManager = new PoolingHttpClientConnectionManager(registry);
connectionManager.setMaxTotal(MAX_TOTAL_CONNECTIONS);
connectionManager.setDefaultMaxPerRoute(MAX_PER_ROUTE_CONNECTIONS);
connectionManager.setValidateAfterInactivity(INACTIVITY_VALIDATED_TIME);
// 构造单例 httpClient
httpClient = HttpClients.custom()
.setConnectionManager(connectionManager)
.setDefaultRequestConfig(requestConfig)
@@ -142,7 +135,6 @@ public class CommonHttpClient {
// 定期清理可用但空闲的连接
.evictIdleConnections(100, TimeUnit.SECONDS)
.build();
// 构造连接清理器
Thread connectCleaner = new Thread(() -> {
while (Thread.currentThread().isInterrupted()) {
try {
@@ -153,7 +145,7 @@ public class CommonHttpClient {
}
}
});
connectCleaner.setName("HttpConnectCleaner");
connectCleaner.setName("http-connection-pool-cleaner");
connectCleaner.setDaemon(true);
connectCleaner.start();
} catch (Exception e) {
@@ -163,8 +155,4 @@ public class CommonHttpClient {
public static CloseableHttpClient getHttpClient() {
return httpClient;
}
public static PoolingHttpClientConnectionManager getConnectionManager() {
return connectionManager;
}
}
@@ -7,22 +7,20 @@ import org.apache.sshd.common.PropertyResolverUtils;
import org.apache.sshd.core.CoreModuleProperties;
/**
* ssh公共client
*
* common ssh pool client
*
*/
@Slf4j
public class CommonSshClient {
private static final SshClient SSH_CLIENT;
static {
SSH_CLIENT = SshClient.setUpDefaultClient();
// 接受所有服务端公钥校验,会打印warn日志 Server at {} presented unverified {} key: {}
// accept all server key verifier, will print warn log : Server at {} presented unverified {} key: {}
AcceptAllServerKeyVerifier verifier = AcceptAllServerKeyVerifier.INSTANCE;
SSH_CLIENT.setServerKeyVerifier(verifier);
// 设置链接保活心跳2000毫秒一次, 客户端等待保活心跳响应超时时间300_000毫秒
// set connection heartbeat interval time 2000ms, wait for heartbeat response timeout 300_000ms
PropertyResolverUtils.updateProperty(
SSH_CLIENT, CoreModuleProperties.HEARTBEAT_INTERVAL.getName(), 2000);
PropertyResolverUtils.updateProperty(
@@ -46,8 +46,7 @@ import java.util.Objects;
import java.util.Optional;
/**
* 数据库JDBC通用查询
*
* common query for database query
*
*/
@Slf4j
@@ -56,15 +55,14 @@ public class JdbcCommonCollect extends AbstractCollect {
private static final String QUERY_TYPE_ONE_ROW = "oneRow";
private static final String QUERY_TYPE_MULTI_ROW = "multiRow";
private static final String QUERY_TYPE_COLUMNS = "columns";
private static final String RUN_SCRIPT = "runScript";
public JdbcCommonCollect(){}
@Override
public void collect(CollectRep.MetricsData.Builder builder, long appId, String app, Metrics metrics) {
public void collect(CollectRep.MetricsData.Builder builder, long monitorId, String app, Metrics metrics) {
long startTime = System.currentTimeMillis();
// 简单校验必有参数
// check the params
if (metrics == null || metrics.getJdbc() == null) {
builder.setCode(CollectRep.Code.FAIL);
builder.setMsg("DATABASE collect must has jdbc params");
@@ -72,7 +70,6 @@ public class JdbcCommonCollect extends AbstractCollect {
}
JdbcProtocol jdbcProtocol = metrics.getJdbc();
String databaseUrl = constructDatabaseUrl(jdbcProtocol);
// 查询超时时间默认6000毫秒
int timeout = CollectUtil.getTimeout(jdbcProtocol.getTimeout());
Statement statement = null;
try {
@@ -147,11 +144,11 @@ public class JdbcCommonCollect extends AbstractCollect {
JdbcConnect jdbcConnect = (JdbcConnect) cacheOption.get();
try {
statement = jdbcConnect.getConnection().createStatement();
// 设置查询超时时间10秒
// set query timeout
int timeoutSecond = timeout / 1000;
timeoutSecond = timeoutSecond <= 0 ? 1 : timeoutSecond;
statement.setQueryTimeout(timeoutSecond);
// 设置查询最大行数1000行
// set query max row number
statement.setMaxRows(1000);
} catch (Exception e) {
log.info("The jdbc connect from cache, create statement error: {}", e.getMessage());
@@ -170,14 +167,12 @@ public class JdbcCommonCollect extends AbstractCollect {
if (statement != null) {
return statement;
}
// 复用失败则新建连接
// renew connection when failed
Connection connection = DriverManager.getConnection(url, username, password);
statement = connection.createStatement();
// 设置查询超时时间10秒
int timeoutSecond = timeout / 1000;
timeoutSecond = timeoutSecond <= 0 ? 1 : timeoutSecond;
statement.setQueryTimeout(timeoutSecond);
// 设置查询最大行数1000行
statement.setMaxRows(1000);
JdbcConnect jdbcConnect = new JdbcConnect(connection);
CommonCache.getInstance().addCache(identifier, jdbcConnect);
@@ -185,13 +180,13 @@ public class JdbcCommonCollect extends AbstractCollect {
}
/**
* 查询一行数据, 通过查询返回结果集的列名称,和查询的字段映射
* query one row record, response metrics header and one value row
* eg:
* 查询字段one tow three four
* 查询SQLselect one, tow, three, four from book limit 1;
* @param statement 执行器
* query metricsone tow three four
* query sqlselect one, tow, three, four from book limit 1;
* @param statement statement
* @param sql sql
* @param columns 查询的列头(一般是数据库表字段,也可能包含特殊字段,eg: responseTime)
* @param columns query metrics field list
* @throws Exception when error happen
*/
private void queryOneRow(Statement statement, String sql, List<String> columns,
@@ -216,14 +211,18 @@ public class JdbcCommonCollect extends AbstractCollect {
}
/**
* 查询一行数据, 通过查询的两列数据(key-value),key和查询的字段匹配,value为查询字段的值
* query two columns to mapping one row
* eg:
* 查询字段one two three four
* 查询SQLselect key, value from book;
* 返回的key映射查询字段
* @param statement 执行器
* query metricsone two three four
* query sqlselect key, value from book; the key is the query metrics fields
* select key, value from book;
* one - value1
* two - value2
* three - value3
* four - value4
* @param statement statement
* @param sql sql
* @param columns 查询的列头(一般是数据库表字段,也可能包含特殊字段,eg: responseTime)
* @param columns query metrics field list
* @throws Exception when error happen
*/
private void queryOneRowByMatchTwoColumns(Statement statement, String sql, List<String> columns,
@@ -251,13 +250,14 @@ public class JdbcCommonCollect extends AbstractCollect {
}
/**
* 查询多行数据, 通过查询返回结果集的列名称,和查询的字段映射
* query multi row record, response metrics header and multi value row
* eg:
* 查询字段one tow three four
* 查询SQLselect one, tow, three, four from book;
* @param statement 执行器
* query metricsone tow three four
* query sqlselect one, tow, three, four from book;
* and return multi row record mapping with the metrics
* @param statement statement
* @param sql sql
* @param columns 查询的列头(一般是数据库表字段,也可能包含特殊字段,eg: responseTime)
* @param columns query metrics field list
* @throws Exception when error happen
*/
private void queryMultiRow(Statement statement, String sql, List<String> columns,
@@ -281,7 +281,7 @@ public class JdbcCommonCollect extends AbstractCollect {
}
/**
* 根据jdbc入参构造数据库URL
* construct jdbc url due the jdbc protocol
* @param jdbcProtocol jdbc
* @return URL
*/
@@ -289,7 +289,7 @@ public class JdbcCommonCollect extends AbstractCollect {
if (Objects.nonNull(jdbcProtocol.getUrl())
&& !Objects.equals("", jdbcProtocol.getUrl())
&& jdbcProtocol.getUrl().startsWith("jdbc")) {
// 入参数URL有效 则优先级最高返回
// when has config jdbc url, use it
return jdbcProtocol.getUrl();
}
String url;
@@ -24,8 +24,7 @@ import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Service;
/**
* 预加载jdbc驱动包 避免spi并发加载造成死锁
*
* load the jdbc driver first to avoid spi concurrent deadlock
*
*/
@Service
@@ -18,9 +18,6 @@ import java.util.Objects;
/**
* ftp protocol collection implementation
* ftp协议采集实现
*
*
*
*/
@Slf4j
@@ -30,7 +27,7 @@ public class FtpCollectImpl extends AbstractCollect {
private final String PASSWORD = "password";
@Override
public void collect(CollectRep.MetricsData.Builder builder, long appId, String app, Metrics metrics) {
public void collect(CollectRep.MetricsData.Builder builder, long monitorId, String app, Metrics metrics) {
FTPClient ftpClient = new FTPClient();
FtpProtocol ftpProtocol = metrics.getFtp();
// Set timeout
@@ -67,10 +64,10 @@ public class FtpCollectImpl extends AbstractCollect {
/**
* collect data: key-value
* Please modify this, if you want to add some indicators.
* Please modify this, if you want to add some metrics.
*/
private Map<String, String> collectValue(FTPClient ftpClient, FtpProtocol ftpProtocol) {
Boolean isActive;
boolean isActive;
String responseTime;
try {
long startTime = System.currentTimeMillis();
@@ -79,14 +76,14 @@ public class FtpCollectImpl extends AbstractCollect {
// In here, we can do some extended operation without changing the architecture
isActive = ftpClient.changeWorkingDirectory(ftpProtocol.getDirection());
long endTime = System.currentTimeMillis();
responseTime = (endTime - startTime) + "";
responseTime = String.valueOf(endTime - startTime);
ftpClient.disconnect();
} catch (Exception e) {
log.info("[FTPClient] error: {}", CommonUtil.getMessageFromThrowable(e), e);
throw new IllegalArgumentException(e.getMessage());
}
return new HashMap<>(8) {{
put("isActive", isActive.toString());
put("isActive", Boolean.toString(isActive));
put("responseTime", responseTime);
}};
}
@@ -101,9 +101,8 @@ public class HttpCollectImpl extends AbstractCollect {
@Override
public void collect(CollectRep.MetricsData.Builder builder,
long appId, String app, Metrics metrics) {
long monitorId, String app, Metrics metrics) {
long startTime = System.currentTimeMillis();
// 校验参数
try {
validateParams(metrics);
} catch (Exception e) {
@@ -120,16 +119,13 @@ public class HttpCollectImpl extends AbstractCollect {
boolean isSuccessInvoke = checkSuccessInvoke(metrics, statusCode);
log.debug("http response status: {}", statusCode);
if (!isSuccessInvoke) {
// 状态码不在successCodes中的状态码为失败
builder.setCode(CollectRep.Code.FAIL);
builder.setMsg("StatusCode " + statusCode);
return;
}
// 在successCodes中的状态码成功
// todo 这里直接将InputStream转为了String, 对于prometheus exporter大数据来说, 会生成大对象, 可能会严重影响JVM内存空间
// todo 方法一、使用InputStream进行解析, 代码改动大; 方法二、手动触发gc, 可以参考dubbo for long i
String resp = EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8);
// 根据不同的解析方式解析
if (resp == null || "".equals(resp)) {
log.info("http response entity is empty, status: {}.", statusCode);
}
@@ -164,25 +160,21 @@ public class HttpCollectImpl extends AbstractCollect {
builder.setCode(CollectRep.Code.UN_CONNECTABLE);
builder.setMsg(errorMsg);
} catch (UnknownHostException e2) {
// 对端不可达
String errorMsg = CommonUtil.getMessageFromThrowable(e2);
log.info(errorMsg);
builder.setCode(CollectRep.Code.UN_REACHABLE);
builder.setMsg("unknown host:" + errorMsg);
} catch (InterruptedIOException | ConnectException | SSLException e3) {
// 对端连接失败
String errorMsg = CommonUtil.getMessageFromThrowable(e3);
log.info(errorMsg);
builder.setCode(CollectRep.Code.UN_CONNECTABLE);
builder.setMsg(errorMsg);
} catch (IOException e4) {
// 其它IO异常
String errorMsg = CommonUtil.getMessageFromThrowable(e4);
log.info(errorMsg);
builder.setCode(CollectRep.Code.FAIL);
builder.setMsg(errorMsg);
} catch (Exception e) {
// 其它异常
String errorMsg = CommonUtil.getMessageFromThrowable(e);
log.error(errorMsg, e);
builder.setCode(CollectRep.Code.FAIL);
@@ -218,7 +210,6 @@ public class HttpCollectImpl extends AbstractCollect {
private void parseResponseByWebsite(String resp, List<String> aliasFields, HttpProtocol http,
CollectRep.MetricsData.Builder builder, Long responseTime) {
CollectRep.ValueRow.Builder valueRowBuilder = CollectRep.ValueRow.newBuilder();
// 网站关键词数量监测
int keywordNum = CollectUtil.countMatchKeyword(resp, http.getKeyword());
for (String alias : aliasFields) {
if (CollectorConstants.RESPONSE_TIME.equalsIgnoreCase(alias)) {
@@ -235,7 +226,6 @@ public class HttpCollectImpl extends AbstractCollect {
private void parseResponseBySiteMap(String resp, List<String> aliasFields,
CollectRep.MetricsData.Builder builder) {
List<String> siteUrls = new LinkedList<>();
// 使用xml解析
boolean isXmlFormat = true;
try {
DocumentBuilderFactory dbf = DocumentBuilderFactory.newInstance();
@@ -288,13 +278,10 @@ public class HttpCollectImpl extends AbstractCollect {
errorMsg = e1.getMessage();
}
} catch (UnknownHostException e2) {
// 对端不可达
errorMsg = "unknown host";
} catch (InterruptedIOException | ConnectException | SSLException e3) {
// 对端连接失败
errorMsg = "connect error: " + e3.getMessage();
} catch (IOException e4) {
// 其它IO异常
errorMsg = "io error: " + e4.getMessage();
} catch (Exception e) {
errorMsg = "error: " + e.getMessage();
@@ -458,7 +445,6 @@ public class HttpCollectImpl extends AbstractCollect {
/**
* create httpContext
*
* @param httpProtocol http protocol
* @return context
*/
@@ -483,14 +469,12 @@ public class HttpCollectImpl extends AbstractCollect {
}
/**
* 根据http配置参数构造请求头
*
* @param httpProtocol http参数配置
* @return 请求体
* create http request
* @param httpProtocol http params
* @return http uri request
*/
public HttpUriRequest createHttpRequest(HttpProtocol httpProtocol) {
RequestBuilder requestBuilder;
// method
String httpMethod = httpProtocol.getMethod().toUpperCase();
if (HttpMethod.GET.matches(httpMethod)) {
requestBuilder = RequestBuilder.get();
@@ -540,11 +524,9 @@ public class HttpCollectImpl extends AbstractCollect {
requestBuilder.addHeader(HttpHeaders.ACCEPT, "*/*");
}
// 判断是否使用Bearer Token认证
if (httpProtocol.getAuthorization() != null) {
HttpProtocol.Authorization authorization = httpProtocol.getAuthorization();
if (DispatchConstants.BEARER_TOKEN.equalsIgnoreCase(authorization.getType())) {
// 若使用 将token放入到header里面
String value = DispatchConstants.BEARER + " " + authorization.getBearerTokenToken();
requestBuilder.addHeader(HttpHeaders.AUTHORIZATION, value);
} else if (DispatchConstants.BASIC_AUTH.equals(authorization.getType())) {
@@ -557,7 +539,7 @@ public class HttpCollectImpl extends AbstractCollect {
}
}
// 请求内容,会覆盖post协议的params
// if it has payload, would override post params
if (StringUtils.hasLength(httpProtocol.getPayload())) {
requestBuilder.setEntity(new StringEntity(httpProtocol.getPayload(), StandardCharsets.UTF_8));
}
@@ -46,7 +46,6 @@ import static org.dromara.hertzbeat.common.constants.SignConstants.RIGHT_DASH;
/**
* ssl Certificate
*
*
*/
@Slf4j
public class SslCertificateCollectImpl extends AbstractCollect {
@@ -62,7 +61,7 @@ public class SslCertificateCollectImpl extends AbstractCollect {
@Override
public void collect(CollectRep.MetricsData.Builder builder,
long appId, String app, Metrics metrics) {
long monitorId, String app, Metrics metrics) {
long startTime = System.currentTimeMillis();
try {
validateParams(metrics);
@@ -126,25 +125,21 @@ public class SslCertificateCollectImpl extends AbstractCollect {
builder.setCode(CollectRep.Code.FAIL);
builder.setMsg(errorMsg);
} catch (UnknownHostException e2) {
// 对端不可达
String errorMsg = CommonUtil.getMessageFromThrowable(e2);
log.info(errorMsg);
builder.setCode(CollectRep.Code.UN_REACHABLE);
builder.setMsg("unknown host:" + errorMsg);
} catch (InterruptedIOException | ConnectException | SSLException e3) {
// 对端连接失败
String errorMsg = CommonUtil.getMessageFromThrowable(e3);
log.info(errorMsg);
builder.setCode(CollectRep.Code.UN_CONNECTABLE);
builder.setMsg(errorMsg);
} catch (IOException e4) {
// 其它IO异常
String errorMsg = CommonUtil.getMessageFromThrowable(e4);
log.info(errorMsg);
builder.setCode(CollectRep.Code.FAIL);
builder.setMsg(errorMsg);
} catch (Exception e) {
// 其它异常
String errorMsg = CommonUtil.getMessageFromThrowable(e);
log.error(errorMsg, e);
builder.setCode(CollectRep.Code.FAIL);
@@ -32,8 +32,7 @@ import java.net.InetAddress;
import java.net.UnknownHostException;
/**
* icmp协议采集实现 - ping
*
* icmp ping collect
*
*/
@Slf4j
@@ -42,7 +41,7 @@ public class IcmpCollectImpl extends AbstractCollect {
public IcmpCollectImpl(){}
@Override
public void collect(CollectRep.MetricsData.Builder builder, long appId, String app, Metrics metrics) {
public void collect(CollectRep.MetricsData.Builder builder, long monitorId, String app, Metrics metrics) {
long startTime = System.currentTimeMillis();
// 简单校验必有参数
if (metrics == null || metrics.getIcmp() == null) {
@@ -59,8 +58,9 @@ public class IcmpCollectImpl extends AbstractCollect {
log.warn(e.getMessage());
}
try {
// todo need root java jcm to use ICMP, else it telnet the peer server 7 port available
// todo 需要配置java虚拟机root权限从而使用ICMP,否则是判断telnet对端7号端口是否开通
// https://stackoverflow.com/questions/11506321/how-to-ping-an-ip-address
// todo https://stackoverflow.com/questions/11506321/how-to-ping-an-ip-address
boolean status = InetAddress.getByName(icmp.getHost()).isReachable(timeout);
long responseTime = System.currentTimeMillis() - startTime;
if (status) {
@@ -26,11 +26,8 @@ import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
/**
* jmx 协议采集实现 - jmx
* jmx protocol acquisition implementation
*
*
*
*/
@Slf4j
public class JmxCollectImpl extends AbstractCollect {
@@ -45,7 +42,7 @@ public class JmxCollectImpl extends AbstractCollect {
}
@Override
public void collect(CollectRep.MetricsData.Builder builder, long appId, String app, Metrics metrics) {
public void collect(CollectRep.MetricsData.Builder builder, long monitorId, String app, Metrics metrics) {
try {
JmxProtocol jmxProtocol = metrics.getJmx();
@@ -44,7 +44,7 @@ import org.dromara.hertzbeat.collector.collect.AbstractCollect;
import lombok.extern.slf4j.Slf4j;
/**
* Mongodb 单机指标收集器
* Mongodb single collect
*
*
* see also https://www.mongodb.com/languages/java,
@@ -80,7 +80,7 @@ public class MongodbSingleCollectImpl extends AbstractCollect {
};
@Override
public void collect(CollectRep.MetricsData.Builder builder, long appId, String app, Metrics metrics) {
public void collect(CollectRep.MetricsData.Builder builder, long monitorId, String app, Metrics metrics) {
try {
preCheck(metrics);
} catch (Exception e) {
@@ -6,10 +6,9 @@ import java.util.List;
import java.util.Map;
/**
* rocketmq采集数据实体类
* rocketmq collect data
*
*
* @since 5/6/2023
*/
@Data
public class RocketmqCollectData {
@@ -48,7 +48,7 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
/**
* rocketmq采集实现类
* rocketmq collect
*
*
* @since 5/6/2023
@@ -96,7 +96,7 @@ public class RocketmqSingleCollectImpl extends AbstractCollect implements Dispos
}
@Override
public void collect(CollectRep.MetricsData.Builder builder, long appId, String app, Metrics metrics) {
public void collect(CollectRep.MetricsData.Builder builder, long monitorId, String app, Metrics metrics) {
try {
preCheck(metrics);
} catch (Exception e) {
@@ -131,9 +131,8 @@ public class RocketmqSingleCollectImpl extends AbstractCollect implements Dispos
}
/**
* 采集前置条件, 入参判断
*
* @param metrics 数据指标
* preCheck params
* @param metrics metrics config
*/
private void preCheck(Metrics metrics) {
if (metrics == null || metrics.getRocketmq() == null) {
@@ -145,9 +144,9 @@ public class RocketmqSingleCollectImpl extends AbstractCollect implements Dispos
}
/**
* 创建DefaultMQAdminExt实体类; 这里有个小问题, 是否需要每次都重新创建
*
* @param metrics 数据指标
* create the DefaultMQAdminExt
* one problem the DefaultMQAdminExt can not reuse
* @param metrics metrics
* @return DefaultMQAdminExt
*/
private DefaultMQAdminExt createMqAdminExt(Metrics metrics) {
@@ -164,11 +163,10 @@ public class RocketmqSingleCollectImpl extends AbstractCollect implements Dispos
}
/**
* 采集rocketmq数据
*
* @param mqAdminExt rocketmq提供的远程调用类
* @param rocketmqCollectData rocketmq数据采集类
* @throws Exception 远程调用异常
* collect rocketmq data
* @param mqAdminExt rocketmq rpc admin
* @param rocketmqCollectData rocketmq data
* @throws Exception when rpc error
*/
private void collectData(DefaultMQAdminExt mqAdminExt, RocketmqCollectData rocketmqCollectData) throws Exception {
this.collectClusterData(mqAdminExt, rocketmqCollectData);
@@ -177,11 +175,10 @@ public class RocketmqSingleCollectImpl extends AbstractCollect implements Dispos
}
/**
* 采集rocketmq的集群数据
*
* @param mqAdminExt rocketmq提供的远程调用类
* @param rocketmqCollectData rocketmq数据采集类
* @throws Exception 远程调用异常
* collect rocketmq cluster data
* @param mqAdminExt rocketmq rpc admin
* @param rocketmqCollectData rocketmq data
* @throws Exception when rpc error
*/
private void collectClusterData(DefaultMQAdminExt mqAdminExt, RocketmqCollectData rocketmqCollectData) throws Exception {
try {
@@ -239,7 +236,6 @@ public class RocketmqSingleCollectImpl extends AbstractCollect implements Dispos
clusterBrokerData.setTodayConsumeCount(todayConsumerCount);
}
}
}
} catch (Exception e) {
log.warn("collect rocketmq cluster data error", e);
@@ -248,16 +244,15 @@ public class RocketmqSingleCollectImpl extends AbstractCollect implements Dispos
}
/**
* 采集rocketmq的消费者数据
*
* @param mqAdminExt rocketmq提供的远程调用类
* @param rocketmqCollectData rocketmq数据采集类
* @throws Exception 远程调用异常
* collect rocketmq consumer data
* @param mqAdminExt rocketmq rpc admin
* @param rocketmqCollectData rocketmq data
* @throws Exception when rpc error
*/
private void collectConsumerData(DefaultMQAdminExt mqAdminExt, RocketmqCollectData rocketmqCollectData) throws Exception {
Set<String> consumerGroupSet = new HashSet<>();
try {
// 获取consumerGroup集合
// get consumerGroup
ClusterInfo clusterInfo = mqAdminExt.examineBrokerClusterInfo();
for (BrokerData brokerData : clusterInfo.getBrokerAddrTable().values()) {
SubscriptionGroupWrapper subscriptionGroupWrapper = mqAdminExt.getAllSubscriptionGroup(brokerData.selectBrokerAddr(), 3000L);
@@ -316,9 +311,10 @@ public class RocketmqSingleCollectImpl extends AbstractCollect implements Dispos
}
/**
* @param mqAdminExt rocketmq提供的远程调用类
* @param rocketmqCollectData rocketmq数据采集类
* @throws Exception 远程调用异常
* collect topic data
* @param mqAdminExt rocketmq rpc admin
* @param rocketmqCollectData rocketmq data
* @throws Exception when rpc error
*/
private void collectTopicData(DefaultMQAdminExt mqAdminExt, RocketmqCollectData rocketmqCollectData) throws Exception {
try {
@@ -333,7 +329,6 @@ public class RocketmqSingleCollectImpl extends AbstractCollect implements Dispos
List<RocketmqCollectData.TopicQueueInfo> topicQueueInfoList = new ArrayList<>();
// todo 查询topic的queue信息需要for循环调用 mqAdminExt.examineTopicStats(), topic数量很大的情况, 调用次数也会很多
topicQueueInfoTable.put(topic, topicQueueInfoList);
topicInfoList.add(topicQueueInfoTable);
rocketmqCollectData.setTopicInfoList(topicInfoList);
@@ -345,12 +340,12 @@ public class RocketmqSingleCollectImpl extends AbstractCollect implements Dispos
}
/**
* 采集数据填充到builder
* fill data to builder
*
* @param rocketmqCollectData rocketmq数据采集类
* @param rocketmqCollectData rocketmq data
* @param builder metrics data builder
* @param aliasFields 字段别名
* @param parseScript JSONbase path
* @param aliasFields alia fields
* @param parseScript JSON base path
*/
private void fillBuilder(RocketmqCollectData rocketmqCollectData, CollectRep.MetricsData.Builder builder, List<String> aliasFields, String parseScript) {
String dataJson = JSONObject.toJSONString(rocketmqCollectData);
@@ -68,8 +68,7 @@ import static org.dromara.hertzbeat.common.constants.SignConstants.RIGHT_DASH;
/**
* http https collect
*
* prometheus auto collect
*
*/
@Slf4j
@@ -100,18 +99,15 @@ public class PrometheusAutoCollectImpl {
boolean isSuccessInvoke = defaultSuccessStatusCodes.contains(statusCode);
log.debug("http response status: {}", statusCode);
if (!isSuccessInvoke) {
// 状态码不在successCodes中的状态码为失败
builder.setCode(CollectRep.Code.FAIL);
builder.setMsg("StatusCode " + statusCode);
return null;
}
// 在successCodes中的状态码成功
// todo 这里直接将InputStream转为了String, 对于prometheus exporter大数据来说, 会生成大对象, 可能会严重影响JVM内存空间
// todo 方法一、使用InputStream进行解析, 代码改动大; 方法二、手动触发gc, 可以参考dubbo for long i
String resp = EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8);
long collectTime = System.currentTimeMillis();
builder.setTime(collectTime);
// 根据不同的解析方式解析
if (resp == null || "".equals(resp)) {
log.error("http response content is empty, status: {}.", statusCode);
builder.setCode(CollectRep.Code.FAIL);
@@ -131,25 +127,21 @@ public class PrometheusAutoCollectImpl {
builder.setCode(CollectRep.Code.UN_CONNECTABLE);
builder.setMsg(errorMsg);
} catch (UnknownHostException e2) {
// 对端不可达
String errorMsg = CommonUtil.getMessageFromThrowable(e2);
log.info(errorMsg);
builder.setCode(CollectRep.Code.UN_REACHABLE);
builder.setMsg("unknown host:" + errorMsg);
} catch (InterruptedIOException | ConnectException | SSLException e3) {
// 对端连接失败
String errorMsg = CommonUtil.getMessageFromThrowable(e3);
log.info(errorMsg);
builder.setCode(CollectRep.Code.UN_CONNECTABLE);
builder.setMsg(errorMsg);
} catch (IOException e4) {
// 其它IO异常
String errorMsg = CommonUtil.getMessageFromThrowable(e4);
log.info(errorMsg);
builder.setCode(CollectRep.Code.FAIL);
builder.setMsg(errorMsg);
} catch (Exception e) {
// 其它异常
String errorMsg = CommonUtil.getMessageFromThrowable(e);
log.error(errorMsg, e);
builder.setCode(CollectRep.Code.FAIL);
@@ -264,12 +256,11 @@ public class PrometheusAutoCollectImpl {
}
return null;
}
/**
* 根据http配置参数构造请求头
*
* @param protocol 参数配置
* @return 请求体
* create http request
* @param protocol http params
* @return http uri request
*/
public HttpUriRequest createHttpRequest(PrometheusProtocol protocol) {
RequestBuilder requestBuilder = RequestBuilder.get();
@@ -299,11 +290,9 @@ public class PrometheusAutoCollectImpl {
// add accept
requestBuilder.addHeader(HttpHeaders.ACCEPT, "*/*");
// 判断是否使用Bearer Token认证
if (protocol.getAuthorization() != null) {
PrometheusProtocol.Authorization authorization = protocol.getAuthorization();
if (DispatchConstants.BEARER_TOKEN.equalsIgnoreCase(authorization.getType())) {
// 若使用 将token放入到header里面
String value = DispatchConstants.BEARER + " " + authorization.getBearerTokenToken();
requestBuilder.addHeader(HttpHeaders.AUTHORIZATION, value);
} else if (DispatchConstants.BASIC_AUTH.equals(authorization.getType())) {
@@ -315,8 +304,8 @@ public class PrometheusAutoCollectImpl {
}
}
}
// 请求内容,会覆盖post协议的params
// if it has payload, would override post params
if (StringUtils.hasLength(protocol.getPayload())) {
requestBuilder.setEntity(new StringEntity(protocol.getPayload(), StandardCharsets.UTF_8));
}
@@ -353,7 +342,7 @@ public class PrometheusAutoCollectImpl {
}
/**
* 获取实例
* get collect instance
* @return instance
*/
public static PrometheusAutoCollectImpl getInstance() {
@@ -361,7 +350,7 @@ public class PrometheusAutoCollectImpl {
}
/**
* 静态内部类
* static instance
*/
private static class SingleInstance {
private static final PrometheusAutoCollectImpl INSTANCE = new PrometheusAutoCollectImpl();
@@ -54,16 +54,16 @@ public class PushCollectImpl extends AbstractCollect {
@Override
public void collect(CollectRep.MetricsData.Builder builder,
long appId, String app, Metrics metrics) {
long monitorId, String app, Metrics metrics) {
long curTime = System.currentTimeMillis();
PushProtocol pushProtocol = metrics.getPush();
Long time = timeMap.getOrDefault(appId, curTime - firstCollectInterval);
timeMap.put(appId, curTime);
Long time = timeMap.getOrDefault(monitorId, curTime - firstCollectInterval);
timeMap.put(monitorId, curTime);
HttpContext httpContext = createHttpContext(pushProtocol);
HttpUriRequest request = createHttpRequest(pushProtocol, appId, time);
HttpUriRequest request = createHttpRequest(pushProtocol, monitorId, time);
try {
CloseableHttpResponse response = CommonHttpClient.getHttpClient().execute(request, httpContext);
@@ -55,7 +55,7 @@ public class RedisCommonCollectImpl extends AbstractCollect {
}
@Override
public void collect(CollectRep.MetricsData.Builder builder, long appId, String app, Metrics metrics) {
public void collect(CollectRep.MetricsData.Builder builder, long monitorId, String app, Metrics metrics) {
try {
preCheck(metrics);
} catch (Exception e) {
@@ -298,9 +298,7 @@ public class RedisCommonCollectImpl extends AbstractCollect {
.forEach(it -> {
result.put(it[0], it[1]);
});
// https://github.com/dromara/hertzbeat/pull/913
// fix 数组越界
// 如果返回的指标数量小于yml配置的指标总和,不区分指标类型 赋值 &nbsp;
// fix https://github.com/dromara/hertzbeat/pull/913
if (result.size() < fieldTotalSize) {
for (Metrics.Field field : metrics.getFields()) {
if (!result.containsKey(field.getField())) {
@@ -47,9 +47,6 @@ import java.util.concurrent.ExecutionException;
/**
* Snmp protocol collection implementation
* snmp 协议采集实现
*
*
*
*/
@Slf4j
@@ -69,9 +66,8 @@ public class SnmpCollectImpl extends AbstractCollect {
@Override
public void collect(CollectRep.MetricsData.Builder builder, long appId, String app, Metrics metrics) {
public void collect(CollectRep.MetricsData.Builder builder, long monitorId, String app, Metrics metrics) {
long startTime = System.currentTimeMillis();
// 校验参数
try {
validateParams(metrics);
} catch (Exception e) {
@@ -60,8 +60,6 @@ import java.util.stream.Collectors;
/**
* Ssh protocol collection implementation
* ssh协议采集实现
*
*
*/
@Slf4j
@@ -77,9 +75,8 @@ public class SshCollectImpl extends AbstractCollect {
}
@Override
public void collect(CollectRep.MetricsData.Builder builder, long appId, String app, Metrics metrics) {
public void collect(CollectRep.MetricsData.Builder builder, long monitorId, String app, Metrics metrics) {
long startTime = System.currentTimeMillis();
// 校验参数
try {
validateParams(metrics);
} catch (Exception e) {
@@ -11,8 +11,6 @@ import java.util.concurrent.ConcurrentHashMap;
/**
* Specific metrics collection factory
* 数据收集策略工厂
*
*
*/
@Configuration
@@ -20,13 +18,12 @@ import java.util.concurrent.ConcurrentHashMap;
public class CollectStrategyFactory implements CommandLineRunner {
/**
* strategy container 策略容器
* strategy container
*/
private static final ConcurrentHashMap<String, AbstractCollect> COLLECT_STRATEGY = new ConcurrentHashMap<>();
/**
* get instance of this protocol collection
* 获取注册的收集实现类
* @param protocol collect protocol
* @return implement of Metrics Collection
*/
@@ -19,6 +19,7 @@ package org.dromara.hertzbeat.collector.collect.telnet;
import org.dromara.hertzbeat.collector.collect.AbstractCollect;
import org.dromara.hertzbeat.collector.dispatch.DispatchConstants;
import org.dromara.hertzbeat.collector.util.CollectUtil;
import org.dromara.hertzbeat.common.constants.CollectorConstants;
import org.dromara.hertzbeat.common.entity.job.Metrics;
import org.dromara.hertzbeat.common.entity.job.protocol.TelnetProtocol;
@@ -35,8 +36,7 @@ import java.util.*;
import java.util.stream.Collectors;
/**
* telnet协议采集实现
*
* telnet collect
*
*/
@Slf4j
@@ -45,9 +45,8 @@ public class TelnetCollectImpl extends AbstractCollect {
public TelnetCollectImpl(){}
@Override
public void collect(CollectRep.MetricsData.Builder builder, long appId, String app, Metrics metrics) {
public void collect(CollectRep.MetricsData.Builder builder, long monitorId, String app, Metrics metrics) {
long startTime = System.currentTimeMillis();
// 简单校验必有参数
if (metrics == null || metrics.getTelnet() == null) {
builder.setCode(CollectRep.Code.FAIL);
builder.setMsg("Telnet collect must has telnet params");
@@ -55,16 +54,9 @@ public class TelnetCollectImpl extends AbstractCollect {
}
TelnetProtocol telnet = metrics.getTelnet();
// 超时时间默认6000毫秒
int timeout = 6000;
try {
timeout = Integer.parseInt(telnet.getTimeout());
} catch (Exception e) {
log.warn(e.getMessage());
}
int timeout = CollectUtil.getTimeout(telnet.getTimeout());
TelnetClient telnetClient = null;
try {
//指明Telnet终端类型,否则会返回来的数据中文会乱码
telnetClient = new TelnetClient("vt200");
telnetClient.setConnectTimeout(timeout);
telnetClient.connect(telnet.getHost(), Integer.parseInt(telnet.getPort()));
@@ -132,7 +124,7 @@ public class TelnetCollectImpl extends AbstractCollect {
String result = new String(telnetClient.getInputStream().readAllBytes());
String[] lines = result.split("\n");
boolean contains = lines[0].contains("=");
Map<String, String> mapValue = Arrays.stream(lines)
return Arrays.stream(lines)
.map(item -> {
if (contains) {
return item.split("=");
@@ -142,6 +134,5 @@ public class TelnetCollectImpl extends AbstractCollect {
})
.filter(item -> item.length == 2)
.collect(Collectors.toMap(x -> x[0], x -> x[1]));
return mapValue;
}
}
@@ -32,8 +32,7 @@ import java.net.*;
import java.nio.charset.StandardCharsets;
/**
* udp探测协议采集实现
*
* udp collect
*
*/
@Slf4j
@@ -45,16 +44,14 @@ public class UdpCollectImpl extends AbstractCollect {
}
@Override
public void collect(CollectRep.MetricsData.Builder builder, long appId, String app, Metrics metrics) {
public void collect(CollectRep.MetricsData.Builder builder, long monitorId, String app, Metrics metrics) {
long startTime = System.currentTimeMillis();
// 简单校验必有参数
if (metrics == null || metrics.getUdp() == null) {
builder.setCode(CollectRep.Code.FAIL);
builder.setMsg("Udp collect must has udp params");
return;
}
UdpProtocol udpProtocol = metrics.getUdp();
// 超时时间默认6000毫秒
int timeout = CollectUtil.getTimeout(udpProtocol.getTimeout());
try (DatagramSocket socket = new DatagramSocket()) {
socket.setSoTimeout(timeout);
@@ -26,27 +26,23 @@ import java.util.List;
/**
* Collection data scheduler interface
* 采集数据调度器接口
*/
public interface CollectDataDispatch {
/**
* Processing and distributing collection result data
* 处理分发采集结果数据
*
* @param timeout time wheel timeout 时间轮timeout
* @param metrics The following indicator group collection tasks 下面的指标组采集任务
* @param metricsData Collect result data 采集结果数据
* @param timeout time wheel timeout
* @param metrics The following metrics collection tasks
* @param metricsData Collect result data
*/
void dispatchCollectData(Timeout timeout, Metrics metrics, CollectRep.MetricsData metricsData);
/**
* Processing and distributing collection result data
* 处理分发采集结果数据
*
* @param timeout time wheel timeout 时间轮timeout
* @param metrics The following indicator group collection tasks 下面的指标组采集任务
* @param metricsDataList Collect result data 采集结果数据
* @param timeout time wheel timeout
* @param metrics The following metrics collection tasks
* @param metricsDataList Collect result data
*/
void dispatchCollectData(Timeout timeout, Metrics metrics, List<CollectRep.MetricsData> metricsDataList);
@@ -47,9 +47,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
/**
* Indicator group collection task and response data scheduler
* 指标组采集任务与响应数据调度器
*
* Collection task and response data scheduler
*
*/
@Component
@@ -57,34 +55,28 @@ import java.util.concurrent.atomic.AtomicReference;
public class CommonDispatcher implements MetricsTaskDispatch, CollectDataDispatch, DisposableBean {
/**
* Metric group collection task timeout value
* 指标组采集任务超时时间值
* Collection task timeout value
*/
private static final long DURATION_TIME = 240_000L;
/**
* trigger sub task max num
* 触发子任务最大数量
* Trigger sub task max num
*/
private static final int MAX_SUB_TASK_NUM = 50;
private static final Gson GSON = new Gson();
/**
* Priority queue of index group collection tasks
* 指标组采集任务优先级队列
* Priority queue of index collection tasks
*/
private final MetricsCollectorQueue jobRequestQueue;
/**
* Time round task scheduler
* 时间轮任务调度器
*/
private final TimerDispatch timerDispatch;
/**
* collection data exporter
* 采集数据导出器
*/
private final CommonDataQueue commonDataQueue;
/**
* Metric group task and start time mapping map
* 指标组任务与开始时间映射map
* Metrics task and start time mapping map
*/
private final Map<String, MetricsTime> metricsTimeoutMonitorMap;
@@ -121,8 +113,7 @@ public class CommonDispatcher implements MetricsTaskDispatch, CollectDataDispatc
public void start() {
try {
// Pull the indicator group collection task from the task queue and put it into the thread pool for execution
// 从任务队列拉取指标组采集任务放入线程池执行
// Pull the collection task from the task queue and put it into the thread pool for execution
poolExecutor.execute(() -> {
Thread.currentThread().setName("metrics-task-dispatcher");
while (!Thread.currentThread().isInterrupted()) {
@@ -138,7 +129,6 @@ public class CommonDispatcher implements MetricsTaskDispatch, CollectDataDispatc
try {
Thread.sleep(1000);
if (metricsCollect != null) {
// 在队列里的优先级增大
metricsCollect.setRunPriority((byte) (metricsCollect.getRunPriority() + 1));
jobRequestQueue.addJob(metricsCollect);
}
@@ -154,19 +144,18 @@ public class CommonDispatcher implements MetricsTaskDispatch, CollectDataDispatc
}
}
});
// Monitoring indicator group collection task execution t
// 监控指标组采集任务执行时间
// Monitoring metrics collection task execution time
poolExecutor.execute(() -> {
Thread.currentThread().setName("metrics-task-monitor");
while (!Thread.currentThread().isInterrupted()) {
try {
// Detect whether the collection unit of each indicator group has timed out for 4 minutes, and if it times out, it will be discarded and an exception will be returned.
// 检测每个指标组采集单元是否超时4分钟,超时则丢弃并返回异常
// Detect whether the collection unit of each metrics has timed out for 4 minutes,
// and if it times out, it will be discarded and an exception will be returned.
long deadline = System.currentTimeMillis() - DURATION_TIME;
for (Map.Entry<String, MetricsTime> entry : metricsTimeoutMonitorMap.entrySet()) {
MetricsTime metricsTime = entry.getValue();
if (metricsTime.getStartTime() < deadline) {
// Metric group collection timeout 指标组采集超时
// Metrics collection timeout
WheelTimerTask timerJob = (WheelTimerTask) metricsTime.getTimeout().task();
CollectRep.MetricsData metricsData = CollectRep.MetricsData.newBuilder()
.setId(timerJob.getJob().getMonitorId())
@@ -199,10 +188,8 @@ public class CommonDispatcher implements MetricsTaskDispatch, CollectDataDispatc
@Override
public void dispatchMetricsTask(Timeout timeout) {
// Divide the collection task of a single application into corresponding collection tasks of the indicator group according to the indicator group under it. AbstractCollect
//Put each indicator group into the thread pool for scheduling
// 将单个应用的采集任务根据其下的指标组拆分为对应的指标组采集任务 AbstractCollect
// 将每个指标组放入线程池进行调度
// Divide the collection task of a single application into corresponding collection tasks of the metrics according to the metrics under it.
// Put each collect task into the thread pool for scheduling
WheelTimerTask timerTask = (WheelTimerTask) timeout.task();
Job job = timerTask.getJob();
job.constructPriorMetrics();
@@ -238,8 +225,7 @@ public class CommonDispatcher implements MetricsTaskDispatch, CollectDataDispatc
}
Set<Metrics> metricsSet = job.getNextCollectMetrics(metrics, false);
if (job.isCyclic()) {
// If it is an asynchronous periodic cyclic task, directly send the collected data of the indicator group to the message middleware
// 若是异步的周期性循环任务,直接发送指标组的采集数据到消息中间件
// If it is an asynchronous periodic cyclic task, directly response the collected data
commonDataQueue.sendMetricsData(metricsData);
if (log.isDebugEnabled()) {
log.debug("Cyclic Job: {} - {} - {}", job.getMonitorId(), job.getApp(), metricsData.getMetrics());
@@ -249,31 +235,25 @@ public class CommonDispatcher implements MetricsTaskDispatch, CollectDataDispatc
}
}
}
//If metricsSet is null, it means that the execution is completed or whether the priority of the collection indicator group is 0, that is, the availability collection indicator group.
// If the availability collection fails, the next indicator group scheduling will be cancelled and the next round of scheduling will be entered directly.
// 若metricsSet为null表示执行完成
// 或判断采集指标组是否优先级为0,即为可用性采集指标组 若可用性采集失败 则取消后面的指标组调度直接进入下一轮调度
// If metricsSet is null, it means that the execution is completed or whether the priority of the collection metrics is 0, that is, the availability collection metrics.
// If the availability collection fails, the next metrics scheduling will be cancelled and the next round of scheduling will be entered directly.
boolean isAvailableCollectFailed = metricsSet != null && !metricsSet.isEmpty()
&& metrics.getPriority() == (byte) 0 && metricsData.getCode() != CollectRep.Code.SUCCESS;
if (metricsSet == null || isAvailableCollectFailed) {
// The collection and execution of all index groups of this job are completed.
// The collection and execution task of this job are completed.
// The periodic task pushes the task to the time wheel again.
// First, determine the execution time of the task and the task collection interval.
// 此Job所有指标组采集执行完成
// 周期性任务再次将任务push到时间轮
// 先判断此次任务执行时间与任务采集间隔时间
if (timeout.isCancelled()) {
return;
}
long spendTime = System.currentTimeMillis() - job.getDispatchTime();
long interval = job.getInterval() - spendTime / 1000;
interval = interval <= 0 ? 0 : interval;
// Reset Construction Execution Metrics Group View 重置构造执行指标组视图
// Reset Construction Execution Metrics Task View
job.constructPriorMetrics();
timerDispatch.cyclicJob(timerJob, interval, TimeUnit.SECONDS);
} else if (!metricsSet.isEmpty()) {
// The execution of the current level indicator group is completed, and the execution of the next level indicator group starts
// 当前级别指标组执行完成,开始执行下一级别的指标组
// The execution of the current level metrics is completed, and the execution of the next level metrics starts
// use pre collect metrics data to replace next metrics config params
List<Map<String, Configmap>> configmapList = getConfigmapFromPreCollectData(metricsData);
for (Metrics metricItem : metricsSet) {
@@ -306,16 +286,12 @@ public class CommonDispatcher implements MetricsTaskDispatch, CollectDataDispatc
}
} else {
// The list of indicator groups at the current execution level has not been fully executed.
// It needs to wait for the execution of other indicator groups of the same level to complete the execution and enter the next level for execution.
// 当前执行级别的指标组列表未全执行完成,
// 需等待其它同级别指标组执行完成后进入下一级别执行
// The list of metrics at the current execution level has not been fully executed.
// It needs to wait for the execution of other metrics task of the same level to complete the execution and enter the next level for execution.
}
} else {
// If it is a temporary one-time task, you need to wait for the collected data of all indicator groups to be packaged and returned.
// Insert the current indicator group data into the job for unified assembly
// 若是临时性一次任务,需等待所有指标组的采集数据统一包装返回
// 将当前指标组数据插入job里统一组装
// If it is a temporary one-time task, you need to wait for the collected data of all metrics task to be packaged and returned.
// Insert the current metrics data into the job for unified assembly
job.addCollectMetricsData(metricsData);
if (log.isDebugEnabled()) {
log.debug("One-time Job: {}", metricsData.getMetrics());
@@ -326,14 +302,11 @@ public class CommonDispatcher implements MetricsTaskDispatch, CollectDataDispatc
}
}
if (metricsSet == null) {
// The collection and execution of all indicator groups of this job are completed
// and the result listener is notified of the combination of all indicator group data
// 此Job所有指标组采集执行完成
// 将所有指标组数据组合一起通知结果监听器
// The collection and execution of all metrics of this job are completed
// and the result listener is notified of the combination of all metrics data
timerDispatch.responseSyncJobData(job.getId(), job.getResponseDataTemp());
} else if (!metricsSet.isEmpty()) {
// The execution of the current level indicator group is completed, and the execution of the next level indicator group starts
// 当前级别指标组执行完成,开始执行下一级别的指标组
// The execution of the current level metrics is completed, and the execution of the next level metrics starts
metricsSet.forEach(metricItem -> {
MetricsCollect metricsCollect = new MetricsCollect(metricItem, timeout, this,
collectorIdentity, unitConvertList);
@@ -342,10 +315,8 @@ public class CommonDispatcher implements MetricsTaskDispatch, CollectDataDispatc
new MetricsTime(System.currentTimeMillis(), metricItem, timeout));
});
} else {
// The list of indicator groups at the current execution level has not been fully executed.
// It needs to wait for the execution of other indicator groups of the same level to complete the execution and enter the next level for execution.
// 当前执行级别的指标组列表未全执行完成,
// 需等待其它同级别指标组执行完成后进入下一级别执行
// The list of metrics task at the current execution level has not been fully executed.
// It needs to wait for the execution of other metrics task of the same level to complete the execution and enter the next level for execution.
}
}
}
@@ -356,27 +327,23 @@ public class CommonDispatcher implements MetricsTaskDispatch, CollectDataDispatc
Job job = timerJob.getJob();
metricsTimeoutMonitorMap.remove(String.valueOf(job.getId()));
if (job.isCyclic()) {
// If it is an asynchronous periodic cyclic task, directly send the collected data of the indicator group to the message middleware
// 若是异步的周期性循环任务,直接发送指标组的采集数据到消息中间件
// If it is an asynchronous periodic cyclic task, directly response the collected data
metricsDataList.forEach(commonDataQueue::sendMetricsData);
// The collection and execution of all index groups of this job are completed.
// The collection and execution of all task of this job are completed.
// The periodic task pushes the task to the time wheel again.
// First, determine the execution time of the task and the task collection interval.
// 此Job所有指标组采集执行完成
// 周期性任务再次将任务push到时间轮
// 先判断此次任务执行时间与任务采集间隔时间
if (timeout.isCancelled()) {
return;
}
long spendTime = System.currentTimeMillis() - job.getDispatchTime();
long interval = job.getInterval() - spendTime / 1000;
interval = interval <= 0 ? 0 : interval;
// Reset Construction Execution Metrics Group View 重置构造执行指标组视图
// Reset Construction Execution Metrics Task View
job.constructPriorMetrics();
timerDispatch.cyclicJob(timerJob, interval, TimeUnit.SECONDS);
} else {
// The collection and execution of all indicator groups of this job are completed
// and the result listener is notified of the combination of all indicator group data
// The collection and execution of all metrics of this job are completed
// and the result listener is notified of the combination of all metrics data
timerDispatch.responseSyncJobData(job.getId(), metricsDataList);
}
@@ -18,7 +18,7 @@
package org.dromara.hertzbeat.collector.dispatch;
/**
* dispatch constant 常量
* dispatch constant
*/
public interface DispatchConstants {
@@ -123,7 +123,7 @@ public interface DispatchConstants {
*/
String PARSE_XML_PATH = "xmlPath";
/**
* Analysis method Website availability monitoring rules Provide responseTime indicators
* Analysis method Website availability monitoring rules Provide responseTime metrics
* 解析方式 网站可用性监控规则 提供responseTime指标
*/
String PARSE_WEBSITE = "website";
@@ -22,7 +22,6 @@ import org.springframework.stereotype.Component;
/**
* Schedule Distribution Task Configuration Properties
* 调度分发任务配置属性
*/
@Component
@ConfigurationProperties(prefix = "collector.dispatch")
@@ -30,13 +29,11 @@ public class DispatchProperties {
/**
* Scheduling entry configuration properties
* 调度入口配置属性
*/
private EntranceProperties entrance;
/**
* Schedule Data Export Configuration Properties
* 调度数据出口配置属性
*/
private ExportProperties export;
@@ -58,10 +55,7 @@ public class DispatchProperties {
/**
* Scheduling entry configuration properties
* The entry can be etcd information, http request, message middleware message request
* <p>
* 调度入口配置属性
* 入口可以时etcd信息,http请求,消息中间件消息请求
* The entry can be netty information, http request, message middleware message request
*/
public static class EntranceProperties {
@@ -69,10 +63,6 @@ public class DispatchProperties {
* netty server client config
*/
private NettyProperties netty;
/**
* etcd配置信息
*/
private EtcdProperties etcd;
public NettyProperties getNetty() {
return netty;
@@ -82,128 +72,7 @@ public class DispatchProperties {
this.netty = netty;
}
public EtcdProperties getEtcd() {
return etcd;
}
public void setEtcd(EtcdProperties etcd) {
this.etcd = etcd;
}
public static class EtcdProperties {
/**
* Whether etcd scheduling is started
* etcd调度是否启动
*/
private boolean enabled = false;
/**
* etcd's connection endpoint url
* etcd的连接端点url
*/
private String[] endpoints = new String[]{"http://127.0.0.1:2379"};
/**
* etcd connection username
* etcd连接用户名
*/
private String username;
/**
* etcd connection password
* etcd连接密码
*/
private String password;
/**
* Valid time of etcd lease in seconds
* etcd租约的有效时间 单位秒
*/
private long ttl = 200;
/**
* Collector registration directory
* 采集器注册目录
*/
private String collectorDir = "/usthe/dispatch/collector/";
/**
* Task scheduling distribution directory
* 任务调度分发目录
*/
private String assignDir = "/usthe/dispatch/assign/";
/**
* task inventory
* 任务详细目录
*/
private String jobDir = "/usthe/dispatch/job/";
public boolean isEnabled() {
return enabled;
}
public void setEnabled(boolean enabled) {
this.enabled = enabled;
}
public String[] getEndpoints() {
return endpoints;
}
public void setEndpoints(String[] endpoints) {
this.endpoints = endpoints;
}
public String getUsername() {
return username;
}
public void setUsername(String username) {
this.username = username;
}
public String getPassword() {
return password;
}
public void setPassword(String password) {
this.password = password;
}
public long getTtl() {
return ttl;
}
public void setTtl(long ttl) {
this.ttl = ttl;
}
public String getCollectorDir() {
return collectorDir;
}
public void setCollectorDir(String collectorDir) {
this.collectorDir = collectorDir;
}
public String getAssignDir() {
return assignDir;
}
public void setAssignDir(String assignDir) {
this.assignDir = assignDir;
}
public String getJobDir() {
return jobDir;
}
public void setJobDir(String jobDir) {
this.jobDir = jobDir;
}
}
public static class NettyProperties {
@@ -39,15 +39,13 @@ import java.util.*;
import java.util.stream.Collectors;
/**
* Index group collection
* 指标组采集
* metrics collection
*/
@Slf4j
@Data
public class MetricsCollect implements Runnable, Comparable<MetricsCollect> {
/**
* Scheduling alarm threshold time 100ms
* 调度告警阈值时间 100ms
*/
private static final long WARN_DISPATCH_TIME = 100;
/**
@@ -60,47 +58,38 @@ public class MetricsCollect implements Runnable, Comparable<MetricsCollect> {
protected long tenantId;
/**
* Monitor ID
* 监控任务ID
*/
protected long monitorId;
/**
* Monitoring type name
* 监控类型名称
*/
protected String app;
/**
* Metric group configuration
* 指标组配置
* Metrics configuration
*/
protected Metrics metrics;
/**
* time wheel timeout
* 时间轮timeout
*/
protected Timeout timeout;
/**
* Task and Data Scheduling
* 任务和数据调度
*/
protected CollectDataDispatch collectDataDispatch;
/**
* task execution priority
* 任务执行优先级
*/
protected byte runPriority;
/**
* Periodic collection or one-time collection true-periodic false-one-time
* 是周期性采集还是一次性采集 true-周期性 false-一次性
*/
protected boolean isCyclic;
/**
* Time for creating an indicator group collection task
* 指标组采集任务新建时间
* Time for creating collection task
*/
protected long newTime;
/**
* Start time of the index group collection task
* 指标组采集任务开始执行时间
* Start time of the collection task
*/
protected long startTime;
@@ -123,7 +112,6 @@ public class MetricsCollect implements Runnable, Comparable<MetricsCollect> {
this.isCyclic = job.isCyclic();
this.unitConvertList = unitConvertList;
// Temporary one-time tasks are executed with high priority
// 临时一次性任务执行优先级高
if (isCyclic) {
runPriority = (byte) -1;
} else {
@@ -148,8 +136,8 @@ public class MetricsCollect implements Runnable, Comparable<MetricsCollect> {
return;
}
response.setMetrics(metrics.getName());
// According to the indicator group collection protocol, application type, etc., dispatch to the real application indicator group collection implementation class
// 根据指标组采集协议,应用类型等来调度到真正的应用指标组采集实现类
// According to the metrics collection protocol, application type, etc.,
// dispatch to the real application metrics collection implementation class
AbstractCollect abstractCollect = CollectStrategyFactory.invoke(metrics.getProtocol());
if (abstractCollect == null) {
log.error("[Dispatcher] - not support this: app: {}, metrics: {}, protocol: {}.",
@@ -173,7 +161,6 @@ public class MetricsCollect implements Runnable, Comparable<MetricsCollect> {
}
}
// Alias attribute expression replacement calculation
// 别名属性表达式替换计算
if (fastFailed()) {
return;
}
@@ -184,14 +171,10 @@ public class MetricsCollect implements Runnable, Comparable<MetricsCollect> {
/**
* Calculate the real indicator (fields) value according to the calculates and aliasFields configuration
* Calculate instance value
* <p>
* 根据 calculates 和 aliasFields 配置计算出真正的指标(fields)值
* 计算instance实例值
* Calculate the real metrics value according to the calculates and aliasFields configuration
*
* @param metrics Metric group configuration 指标组配置
* @param collectData Data collection 采集数据
* @param metrics Metrics configuration
* @param collectData Data collection
*/
private void calculateFields(Metrics metrics, CollectRep.MetricsData.Builder collectData) {
collectData.setPriority(metrics.getPriority());
@@ -210,11 +193,11 @@ public class MetricsCollect implements Runnable, Comparable<MetricsCollect> {
return;
}
collectData.clearValues();
// Preprocess calculates first 先预处理 calculates
// Preprocess calculates first
if (metrics.getCalculates() == null) {
metrics.setCalculates(Collections.emptyList());
}
// eg: database_pages=Database pages unconventional mapping 非常规映射
// eg: database_pages=Database pages unconventional mapping
Map<String, String> fieldAliasMap = new HashMap<>(8);
Map<String, Expression> fieldExpressionMap = metrics.getCalculates()
.stream()
@@ -244,7 +227,6 @@ public class MetricsCollect implements Runnable, Comparable<MetricsCollect> {
}
}
StringBuilder instanceBuilder = new StringBuilder();
for (Metrics.Field field : fields) {
String realField = field.getField();
Expression expression = fieldExpressionMap.get(realField);
@@ -252,7 +234,6 @@ public class MetricsCollect implements Runnable, Comparable<MetricsCollect> {
String aliasFieldUnit = null;
if (expression != null) {
// If there is a calculation expression, calculate the value
// 存在计算表达式 则计算值
if (CommonConstants.TYPE_NUMBER == field.getType()) {
for (String variable : expression.getVariableFullNames()) {
// extract double value and unit from aliasField value
@@ -283,7 +264,6 @@ public class MetricsCollect implements Runnable, Comparable<MetricsCollect> {
}
} else {
// does not exist then map the alias value
// 不存在 则映射别名值
String aliasField = fieldAliasMap.get(realField);
if (aliasField != null) {
value = aliasFieldValueMap.get(aliasField);
@@ -306,8 +286,7 @@ public class MetricsCollect implements Runnable, Comparable<MetricsCollect> {
}
}
}
// 单位处理
Pair<String, String> unitPair = fieldUnitMap.get(realField);
if (aliasFieldUnit != null) {
if (unitPair != null) {
@@ -323,8 +302,7 @@ public class MetricsCollect implements Runnable, Comparable<MetricsCollect> {
}
}
}
// Handle indicator values that may have units such as 34%, 34Mb, and limit values to 4 decimal places
// 处理可能带单位的指标数值 比如 34%, 34Mb,并将数值小数点限制到4位
// Handle metrics values that may have units such as 34%, 34Mb, and limit values to 4 decimal places
if (CommonConstants.TYPE_NUMBER == field.getType()) {
value = CommonUtil.parseDoubleStr(value, field.getUnit());
}
@@ -342,9 +320,9 @@ public class MetricsCollect implements Runnable, Comparable<MetricsCollect> {
/**
* @param cal
* @param fieldAliasMap
* @return
* @param cal cal
* @param fieldAliasMap field alias map
* @return expr
*/
private Object[] transformCal(String cal, Map<String, String> fieldAliasMap) {
int splitIndex = cal.indexOf("=");
@@ -363,9 +341,8 @@ public class MetricsCollect implements Runnable, Comparable<MetricsCollect> {
/**
* transform unit
*
* @param unit
* @return
* @param unit unit
* @return units
*/
private Object[] transformUnit(String unit) {
int equalIndex = unit.indexOf("=");
@@ -25,7 +25,6 @@ import java.util.concurrent.TimeUnit;
/**
* queue of jobs to run
* 待运行的job队列
*/
@Component
@Slf4j
@@ -20,14 +20,12 @@ package org.dromara.hertzbeat.collector.dispatch;
import org.dromara.hertzbeat.collector.dispatch.timer.Timeout;
/**
* Metric group collection task scheduler interface
* 指标组采集任务调度器接口
* Metrics collection task scheduler interface
*/
public interface MetricsTaskDispatch {
/**
* schedule 调度
*
* schedule task
* @param timeout timeout
*/
void dispatchMetricsTask(Timeout timeout);
@@ -30,7 +30,6 @@ import java.util.concurrent.TimeUnit;
/**
* Collection task worker thread pool
* 采集任务工作线程池
*/
@Component
@Slf4j
@@ -43,7 +42,7 @@ public class WorkerPool implements DisposableBean {
}
private void initWorkExecutor() {
// thread factory 线程工厂
// thread factory
ThreadFactory threadFactory = new ThreadFactoryBuilder()
.setUncaughtExceptionHandler((thread, throwable) -> {
log.error("workerExecutor has uncaughtException.");
@@ -63,10 +62,9 @@ public class WorkerPool implements DisposableBean {
/**
* Run the collection task thread
* 运行采集任务线程
*
* @param runnable Task 任务
* @throws RejectedExecutionException when thread pool full 线程池满
* @param runnable Task
* @throws RejectedExecutionException when thread pool full
*/
public void executeJob(Runnable runnable) throws RejectedExecutionException {
workerExecutor.execute(runnable);
@@ -429,7 +429,7 @@ public class HashedWheelTimer implements Timer {
// Initialize the startTime.
startTime = System.nanoTime();
if (startTime == 0) {
// We use 0 as an indicator for the uninitialized value here, so make sure it's not 0 when initialized.
// We use 0 as an metric for the uninitialized value here, so make sure it's not 0 when initialized.
startTime = 1;
}
@@ -26,8 +26,7 @@ import java.util.List;
import java.util.concurrent.TimeUnit;
/**
* 时间轮调度接口
*
* timer dispatch service
*
*
*/
@@ -35,29 +34,24 @@ public interface TimerDispatch {
/**
* Add new job
* 增加新的job
*
* @param addJob job
* @param eventListener One-time synchronous task listener, asynchronous task does not need listener一次性同步任务监听器,异步任务不需要listener
* @param eventListener One-time synchronous task listener, asynchronous task does not need listener
*/
void addJob(Job addJob, CollectResponseEventListener eventListener);
/**
* 调度循环周期性job
*
* Cyclic job
* @param timerTask timerTask
* @param interval 开始调度的间隔时间
* @param timeUnit 时间单位
* @param interval collect interval
* @param timeUnit time unit
*/
void cyclicJob(WheelTimerTask timerTask, long interval, TimeUnit timeUnit);
/**
* Delete existing job
* 删除存在的job
*
* @param jobId jobId
* @param isCyclic Whether it is a periodic task, true is, false is a temporary task
* 是否是周期性任务,true是, false为临时性任务
*/
void deleteJob(long jobId, boolean isCyclic);
@@ -72,10 +66,9 @@ public interface TimerDispatch {
void goOffline();
/**
* 一次性同步采集任务采集结果通知监听器
*
* response sync collect task data
* @param jobId jobId
* @param metricsDataTemps 采集结果数据
* @param metricsDataTemps collect data
*/
void responseSyncJobData(long jobId, List<CollectRep.MetricsData> metricsDataTemps);
}
@@ -40,22 +40,18 @@ public class TimerDispatcher implements TimerDispatch, DisposableBean {
/**
* time round schedule
* 时间轮调度
*/
private final Timer wheelTimer;
/**
* Existing periodic scheduled tasks
* 已存在的周期性调度任务
*/
private final Map<Long, Timeout> currentCyclicTaskMap;
/**
* Existing temporary scheduled tasks
* 已存在的临时性调度任务
*/
private final Map<Long, Timeout> currentTempTaskMap;
/**
* One-time task response listener holds
* 一次性任务响应监听器持有
* jobId - listener
*/
private final Map<Long, CollectResponseEventListener> eventListeners;
@@ -101,7 +97,7 @@ public class TimerDispatcher implements TimerDispatch, DisposableBean {
return;
}
Long jobId = timerTask.getJob().getId();
// 判断此周期性job是否已经被取消
// whether is the job has been canceled
if (currentCyclicTaskMap.containsKey(jobId)) {
Timeout timeout = wheelTimer.newTimeout(timerTask, interval, TimeUnit.SECONDS);
currentCyclicTaskMap.put(timerTask.getJob().getId(), timeout);
@@ -27,7 +27,6 @@ import org.dromara.hertzbeat.common.constants.CommonConstants;
import org.dromara.hertzbeat.common.entity.job.Configmap;
import org.dromara.hertzbeat.common.entity.job.Job;
import org.dromara.hertzbeat.common.entity.job.Metrics;
import org.dromara.hertzbeat.common.entity.job.protocol.PushProtocol;
import org.dromara.hertzbeat.common.support.SpringContextHolder;
import org.dromara.hertzbeat.common.util.AesUtil;
@@ -38,8 +37,6 @@ import java.util.stream.Collectors;
/**
* Timer Task implementation
* TimerTask实现
*
*
*/
@Slf4j
@@ -53,22 +50,18 @@ public class WheelTimerTask implements TimerTask {
this.metricsTaskDispatch = SpringContextHolder.getBean(MetricsTaskDispatch.class);
this.job = job;
// The initialization job will monitor the actual parameter value and replace the collection field
// 初始化job 将监控实际参数值对采集字段进行替换
initJobMetrics(job);
}
/**
* Initialize job fill information
* 初始化job填充信息
*
* @param job job
*/
private void initJobMetrics(Job job) {
// 将监控实际参数值对采集字段进行替换
List<Configmap> config = job.getConfigmap();
Map<String, Configmap> configmap = config.stream()
.peek(item -> {
// 对加密串进行解密
// decode password
if (item.getType() == CommonConstants.PARAM_TYPE_PASSWORD && item.getValue() != null) {
String decodeValue = AesUtil.aesDecode(String.valueOf(item.getValue()));
if (decodeValue == null) {
@@ -1,10 +1,7 @@
package org.dromara.hertzbeat.collector.dispatch.unit;
import java.util.concurrent.TimeUnit;
/**
* the enum of time length
* 时间长短的枚举类
*
*
*/
@@ -33,8 +33,7 @@ import java.util.regex.Matcher;
import java.util.regex.Pattern;
/**
* 采集器工具类
*
* util for collector
*
*/
@Slf4j
@@ -52,11 +51,11 @@ public class CollectUtil {
private static final List<String> UNIT_SYMBOLS = Arrays.asList("%", "G", "g", "M", "m", "K", "k", "B", "b");
/**
* 关键字匹配计数
* count match keyword number
*
* @param content 内容
* @param keyword 关键字
* @return 匹配次数
* @param content content
* @param keyword keyword
* @return match num
*/
public static int countMatchKeyword(String content, String keyword) {
if (content == null || "".equals(content) || keyword == null || "".equals(keyword.trim())) {
@@ -194,7 +193,6 @@ public class CollectUtil {
/**
* json parameter replacement
* json 参数替换
*
* @param jsonElement json
* @param configmap parameter map
@@ -279,7 +277,6 @@ public class CollectUtil {
/**
* json parameter replacement
* json 参数替换
*
* @param jsonElement json
* @param configmap parameter map
@@ -294,7 +291,6 @@ public class CollectUtil {
JsonElement element = entry.getValue();
String key = entry.getKey();
// Replace the attributes of the KEY-VALUE case such as http headers params
// 替换KEY-VALUE情况的属性 比如http headers params
if (key != null && key.startsWith(SMILING_PLACEHOLDER) && key.endsWith(SMILING_PLACEHOLDER)) {
key = key.replaceAll(SMILING_PLACEHOLDER_REX, "");
Configmap param = configmap.get(key);
@@ -315,10 +311,8 @@ public class CollectUtil {
continue;
}
// Replace normal VALUE value
// 替换正常的VALUE值
if (element.isJsonPrimitive()) {
// Check if there are special characters Replace
// 判断是否含有特殊字符 替换
String value = element.getAsString();
Matcher smilingMatcher = SMILING_PLACEHOLDER_REGEX_PATTERN.matcher(value);
if (smilingMatcher.find()) {
@@ -353,7 +347,6 @@ public class CollectUtil {
JsonElement element = jsonArray.get(index);
if (element.isJsonPrimitive()) {
// Check if there are special characters Replace
// 判断是否含有特殊字符 替换
String value = element.getAsString();
Matcher smilingMatcher = SMILING_PLACEHOLDER_REGEX_PATTERN.matcher(value);
if (smilingMatcher.find()) {
@@ -416,10 +409,10 @@ public class CollectUtil {
}
/**
* 将16进制字符串转换为byte[]
* convert 16 hexString to byte[]
* eg: 302c0201010409636f6d6d756e697479a11c020419e502e7020100020100300e300c06082b060102010102000500
* 16进制字符串不区分大小写,返回的数组相同
* @param hexString 16进制字符串
* @param hexString 16 hexString
* @return byte[]
*/
public static byte[] fromHexString(String hexString) {
@@ -427,12 +420,9 @@ public class CollectUtil {
return null;
}
byte[] bytes = new byte[hexString.length() / HEX_STR_WIDTH];
// 16进制字符串
String hex;
for (int i = 0; i < hexString.length() / HEX_STR_WIDTH; i++) {
// 每次截取2位
hex = hexString.substring(i * HEX_STR_WIDTH, i * HEX_STR_WIDTH + HEX_STR_WIDTH);
// 16进制 --> 十进制
bytes[i] = (byte) Integer.parseInt(hex, 16);
}
return bytes;
@@ -26,7 +26,6 @@ import java.util.*;
/**
* json path parser
*
*
*/
public class JsonPathParser {
@@ -41,10 +40,10 @@ public class JsonPathParser {
}
/**
* 使用jsonPath来解析json内容
* @param content json内容
* @param jsonPath jsonPath脚本
* @return 解析后的内容 [{'name': 'tom', 'speed': '433'},{'name': 'lili', 'speed': '543'}]
* use json path to parse content
* @param content json content
* @param jsonPath jsonPath
* @return content [{'name': 'tom', 'speed': '433'},{'name': 'lili', 'speed': '543'}]
*/
public static List<Object> parseContentWithJsonPath(String content, String jsonPath) {
if (content == null || jsonPath == null || "".equals(content) || "".equals(jsonPath)) {
@@ -54,10 +53,10 @@ public class JsonPathParser {
}
/**
* 使用jsonPath来解析json内容
* @param content json内容
* @param jsonPath jsonPath脚本
* @return 解析后的内容 [{'name': 'tom', 'speed': '433'},{'name': 'lili', 'speed': '543'}]
* use json path to parse content
* @param content json content
* @param jsonPath jsonPath
* @return content [{'name': 'tom', 'speed': '433'},{'name': 'lili', 'speed': '543'}]
*/
public static <T> T parseContentWithJsonPath(String content, String jsonPath, TypeRef<T> typeRef) {
if (content == null || jsonPath == null || "".equals(content) || "".equals(jsonPath)) {
@@ -9,10 +9,9 @@ import java.nio.file.Paths;
import java.util.Objects;
/**
* 将私钥写入~/.ssh
* private key util
* write private key to ~/.ssh
*
*
* Created by gcdd1993 on 2023/7/9
*/
@Slf4j
@UtilityClass
@@ -20,7 +20,7 @@ package org.dromara.hertzbeat.common.cache;
import java.time.Duration;
/**
*
* common cache factory
*
*/
public class CacheFactory {
@@ -36,7 +36,7 @@ public class CacheFactory {
new CaffeineCacheServiceImpl<>(10, 1000, Duration.ofDays(1), false);
/**
* 获取notice模块的cache
* get notice cache
* @return caffeine cache
*/
public static ICacheService<String, Object> getNoticeCache() {
@@ -23,7 +23,7 @@ import com.github.benmanes.caffeine.cache.Caffeine;
import java.time.Duration;
/**
*
* caffeine cache impl
*
*/
public class CaffeineCacheServiceImpl<K, V> implements ICacheService<K, V> {
@@ -18,8 +18,7 @@
package org.dromara.hertzbeat.common.cache;
/**
* common cache
*
* common cache service
*
*/
public interface ICacheService<K, V> {
@@ -34,7 +34,7 @@ import java.util.Objects;
import java.util.regex.Pattern;
/**
*
* aviator config
*
*/
@Configuration
@@ -23,7 +23,7 @@ import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
/**
*
* common module config
*
*/
@ComponentScan(basePackages = "org.dromara.hertzbeat.common")
@@ -20,8 +20,7 @@ package org.dromara.hertzbeat.common.config;
import org.springframework.boot.context.properties.ConfigurationProperties;
/**
* common properties
*
* common module properties
*
*
*/
@@ -18,8 +18,7 @@
package org.dromara.hertzbeat.common.constants;
/**
* collector 常量
*
* collector constant
*
*/
public interface CollectorConstants {
@@ -39,7 +38,7 @@ public interface CollectorConstants {
String HTTPS_HEADER = "https://";
/**
* POSTGRESQL状态码 不可达
* POSTGRESQL un reachable status code
*/
String POSTGRESQL_UN_REACHABLE_CODE = "08001";
@@ -18,8 +18,7 @@
package org.dromara.hertzbeat.common.constants;
/**
* Public Constant
*
* Public Common Constant
*
*/
public interface CommonConstants {
@@ -158,7 +157,7 @@ public interface CommonConstants {
byte TYPE_TIME = 3;
/**
* Collection indicator value: null placeholder for empty value
* Collection metric value: null placeholder for empty value
* 采集指标值:null空值占位符
*/
String NULL_VALUE = "&nbsp;";
@@ -1,8 +1,7 @@
package org.dromara.hertzbeat.common.constants;
/**
* 特殊字符常量
*
* Sign Constants
*
*/
public interface SignConstants {
@@ -17,5 +16,4 @@ public interface SignConstants {
String CARRIAGE_RETURN = "\r";
String RIGHT_DASH = "/";
}
@@ -41,9 +41,7 @@ import static io.swagger.v3.oas.annotations.media.Schema.AccessMode.READ_ONLY;
import static io.swagger.v3.oas.annotations.media.Schema.AccessMode.READ_WRITE;
/**
* Alarm record entity 告警记录实体
*
*
* Alarm record entity
*
*/
@Entity
@@ -74,8 +72,7 @@ public class Alert {
example = "8743267443543", accessMode = READ_WRITE)
private Long alertDefineId;
@Schema(title = "Alarm level 0: high-emergency-critical alarm-red 1: medium-critical-critical alarm-orange 2: low-warning-warning alarm-yellow",
description = "告警级别 0:高-emergency-紧急告警-红色 1:中-critical-严重告警-橙色 2:低-warning-警告告警-黄色",
@Schema(title = "Alarm level 0:High-Emergency-Critical Alarm 1:Medium-Critical-Critical Alarm 2:Low-Warning-Warning",
example = "1", accessMode = READ_WRITE)
@Min(0)
@Max(2)
@@ -122,11 +119,11 @@ public class Alert {
@Column(length = 2048)
private Map<String, String> tags;
@Schema(title = "此条记录创建者", example = "tom", accessMode = READ_ONLY)
@Schema(title = "The creator of this record", example = "tom", accessMode = READ_ONLY)
@CreatedBy
private String creator;
@Schema(title = "此条记录最新修改者", example = "tom", accessMode = READ_ONLY)
@Schema(title = "The modifier of this record", example = "tom", accessMode = READ_ONLY)
@LastModifiedBy
private String modifier;
@@ -136,7 +133,7 @@ public class Alert {
@CreatedDate
private LocalDateTime gmtCreate;
@Schema(title = "记录最新修改时间", example = "1612198444000", accessMode = READ_ONLY)
@Schema(title = "Record modify time", example = "1612198444000", accessMode = READ_ONLY)
@LastModifiedDate
private LocalDateTime gmtUpdate;
@@ -43,8 +43,6 @@ import static io.swagger.v3.oas.annotations.media.Schema.AccessMode.READ_WRITE;
/**
* Alert Converge strategy entity
* 告警收敛策略
*
*
*/
@Entity
@@ -78,7 +76,7 @@ public class AlertConverge {
example = "true", accessMode = READ_WRITE)
private boolean matchAll = true;
@Schema(title = "匹配告警级别,空为全部告警级别 0:高-emergency-紧急告警-红色 1:中-critical-严重告警-橙色 2:低-warning-警告告警-黄色",
@Schema(title = "Alarm Level 0:High-Emergency-Critical Alarm 1:Medium-Critical-Critical Alarm 2:Low-Warning-Warning",
example = "[1]", accessMode = READ_WRITE)
@Convert(converter = JsonByteListAttributeConverter.class)
private List<Byte> priorities;
@@ -93,23 +91,19 @@ public class AlertConverge {
@Min(0)
private Integer evalInterval;
@Schema(title = "The creator of this record", description = "此条记录创建者", example = "tom", accessMode = READ_ONLY)
@Schema(title = "The creator of this record", example = "tom", accessMode = READ_ONLY)
@CreatedBy
private String creator;
@Schema(title = "This record was last modified by",
description = "此条记录最新修改者",
example = "tom", accessMode = READ_ONLY)
@Schema(title = "This record was last modified by", example = "tom", accessMode = READ_ONLY)
@LastModifiedBy
private String modifier;
@Schema(title = "This record creation time (millisecond timestamp)",
description = "记录创建时间", accessMode = READ_ONLY)
@Schema(title = "This record creation time (millisecond timestamp)", accessMode = READ_ONLY)
@CreatedDate
private LocalDateTime gmtCreate;
@Schema(title = "Record the latest modification time (timestamp in milliseconds)",
description = "记录最新修改时间", accessMode = READ_ONLY)
@Schema(title = "Record the latest modification time (timestamp in milliseconds)", accessMode = READ_ONLY)
@LastModifiedDate
private LocalDateTime gmtUpdate;
}
@@ -45,8 +45,7 @@ import static io.swagger.v3.oas.annotations.media.Schema.AccessMode.READ_WRITE;
/**
* 告警定义实体
*
* Alarm Define Rule Entity
*
*/
@Entity
@@ -55,80 +54,80 @@ import static io.swagger.v3.oas.annotations.media.Schema.AccessMode.READ_WRITE;
@Builder
@AllArgsConstructor
@NoArgsConstructor
@Schema(description = "Alert Define | 告警定义实体")
@Schema(description = "Alarm Threshold Entity | 告警阈值实体")
@EntityListeners(AuditingEntityListener.class)
public class AlertDefine {
@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
@Schema(title = "告警定义实体主键索引ID", example = "87584674384", accessMode = READ_ONLY)
@Schema(title = "Threshold Id", example = "87584674384", accessMode = READ_ONLY)
private Long id;
@Schema(title = "配置告警的监控类型", example = "linux", accessMode = READ_WRITE)
@Schema(title = "Monitoring Type", example = "linux", accessMode = READ_WRITE)
@Length(max = 100)
@NotNull
private String app;
@Schema(title = "配置告警的指标集合", example = "cpu", accessMode = READ_WRITE)
@Schema(title = "Monitoring Metrics", example = "cpu", accessMode = READ_WRITE)
@Length(max = 100)
@NotNull
private String metric;
@Schema(title = "配置告警的指标", example = "usage", accessMode = READ_WRITE)
@Schema(title = "Monitoring Metrics Field", example = "usage", accessMode = READ_WRITE)
@Length(max = 100)
private String field;
@Schema(title = "是否是全局默认告警", example = "false", accessMode = READ_WRITE)
@Schema(title = "Is Apply All Default | 是否是全局默认告警", example = "false", accessMode = READ_WRITE)
private boolean preset;
@Schema(title = "告警阈值触发条件表达式", example = "usage>90", accessMode = READ_WRITE)
@Schema(title = "Alarm Threshold Expr | 告警阈值触发条件表达式", example = "usage>90", accessMode = READ_WRITE)
@Length(max = 2048)
@Column(length = 2048)
private String expr;
@Schema(title = "告警级别 0:高-emergency-紧急告警-红色 1:中-critical-严重告警-橙色 2:低-warning-警告告警-黄色",
@Schema(title = "Alarm Level 0:High-Emergency-Critical Alarm 1:Medium-Critical-Critical Alarm 2:Low-Warning-Warning",
example = "1", accessMode = READ_WRITE)
@Min(0)
@Max(2)
private byte priority;
@Schema(title = "阈值触发次数,即达到次数要求后才触发告警", example = "3", accessMode = READ_WRITE)
@Schema(title = "Alarm Trigger Times | 阈值触发次数,即达到次数要求后才触发告警", example = "3", accessMode = READ_WRITE)
@Min(0)
@Max(10)
private Integer times;
@Schema(description = "附加告警标签(status:success,env:prod)", example = "{name: key1, value: value1}",
@Schema(description = "Tags(status:success,env:prod)", example = "{name: key1, value: value1}",
accessMode = READ_WRITE)
@Convert(converter = JsonTagListAttributeConverter.class)
@Column(length = 2048)
private List<TagItem> tags;
@Schema(title = "告警阈值开关", example = "true", accessMode = READ_WRITE)
@Schema(title = "Is Enable", example = "true", accessMode = READ_WRITE)
private boolean enable = true;
@Schema(title = "Is send alarm recover notice | 是否发送告警恢复通知", example = "false", accessMode = READ_WRITE)
@Schema(title = "Is Send Alarm Recover Notice | 是否发送告警恢复通知", example = "false", accessMode = READ_WRITE)
@Column(columnDefinition = "boolean default false")
private boolean recoverNotice = false;
@Schema(title = "告警通知内容模版", example = "linux {monitor_name}: {monitor_id} cpu usage high",
@Schema(title = "Alarm Template | 告警通知内容模版", example = "linux {monitor_name}: {monitor_id} cpu usage high",
accessMode = READ_WRITE)
@Length(max = 2048)
@Column(length = 2048)
private String template;
@Schema(title = "此条记录创建者", example = "tom", accessMode = READ_ONLY)
@Schema(title = "The creator of this record", example = "tom", accessMode = READ_ONLY)
@CreatedBy
private String creator;
@Schema(title = "此条记录最新修改者", example = "tom", accessMode = READ_ONLY)
@Schema(title = "The modifier of this record", example = "tom", accessMode = READ_ONLY)
@LastModifiedBy
private String modifier;
@Schema(title = "记录创建时间", example = "1612198922000", accessMode = READ_ONLY)
@Schema(title = "Record create time", example = "1612198922000", accessMode = READ_ONLY)
@CreatedDate
private LocalDateTime gmtCreate;
@Schema(title = "记录最新修改时间", example = "1612198444000", accessMode = READ_ONLY)
@Schema(title = "Record modify time", example = "1612198444000", accessMode = READ_ONLY)
@LastModifiedDate
private LocalDateTime gmtUpdate;
@@ -37,8 +37,7 @@ import static io.swagger.v3.oas.annotations.media.Schema.AccessMode.READ_WRITE;
/**
* 告警定义与监控关联实体
*
* Alarm Threshold Relate Monitor Entity
*
*/
@Entity
@@ -50,27 +49,27 @@ import static io.swagger.v3.oas.annotations.media.Schema.AccessMode.READ_WRITE;
@Builder
@AllArgsConstructor
@NoArgsConstructor
@Schema(description = "告警定义与监控关联实体")
@Schema(description = "Alarm Threshold Relate Monitor Entity")
@EntityListeners(AuditingEntityListener.class)
public class AlertDefineMonitorBind {
@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
@Schema(title = "告警定义与监控关联主键索引ID", example = "74384", accessMode = READ_ONLY)
@Schema(title = "id", example = "74384", accessMode = READ_ONLY)
private Long id;
@Schema(title = "告警定义ID", example = "87432674384", accessMode = READ_WRITE)
@Schema(title = "Alarm Define Id", example = "87432674384", accessMode = READ_WRITE)
private Long alertDefineId;
@Schema(title = "监控任务ID", example = "87432674336", accessMode = READ_WRITE)
@Schema(title = "Monitor Id", example = "87432674336", accessMode = READ_WRITE)
@Column(name = "monitor_id")
private Long monitorId;
@Schema(title = "记录创建时间", example = "1612198922000", accessMode = READ_ONLY)
@Schema(title = "Record create time", example = "1612198922000", accessMode = READ_ONLY)
@CreatedDate
private LocalDateTime gmtCreate;
@Schema(title = "记录最新修改时间", example = "1612198444000", accessMode = READ_ONLY)
@Schema(title = "Record modify time", example = "1612198444000", accessMode = READ_ONLY)
@LastModifiedDate
private LocalDateTime gmtUpdate;
@@ -43,8 +43,6 @@ import static io.swagger.v3.oas.annotations.media.Schema.AccessMode.READ_WRITE;
/**
* Alert Silence strategy entity
* 告警静默策略
*
*
*/
@Entity
@@ -90,7 +88,7 @@ public class AlertSilence {
description = "已静默告警次数", accessMode = READ_WRITE)
private Integer times;
@Schema(title = "匹配告警级别,空为全部告警级别 0:高-emergency-紧急告警-红色 1:中-critical-严重告警-橙色 2:低-warning-警告告警-黄色",
@Schema(title = "Alarm Level 0:High-Emergency-Critical Alarm 1:Medium-Critical-Critical Alarm 2:Low-Warning-Warning",
example = "[1]", accessMode = READ_WRITE)
@Convert(converter = JsonByteListAttributeConverter.class)
private List<Byte> priorities;
@@ -111,23 +109,19 @@ public class AlertSilence {
@Schema(title = "限制时间段截止", example = "23:59:59", accessMode = READ_WRITE)
private ZonedDateTime periodEnd;
@Schema(title = "The creator of this record", description = "此条记录创建者", example = "tom", accessMode = READ_ONLY)
@Schema(title = "The creator of this record", example = "tom", accessMode = READ_ONLY)
@CreatedBy
private String creator;
@Schema(title = "This record was last modified by",
description = "此条记录最新修改者",
example = "tom", accessMode = READ_ONLY)
@Schema(title = "This record was last modified by", example = "tom", accessMode = READ_ONLY)
@LastModifiedBy
private String modifier;
@Schema(title = "This record creation time (millisecond timestamp)",
description = "记录创建时间", accessMode = READ_ONLY)
@Schema(title = "This record creation time (millisecond timestamp)", accessMode = READ_ONLY)
@CreatedDate
private LocalDateTime gmtCreate;
@Schema(title = "Record the latest modification time (timestamp in milliseconds)",
description = "记录最新修改时间", accessMode = READ_ONLY)
@Schema(title = "Record the latest modification time (timestamp in milliseconds)", accessMode = READ_ONLY)
@LastModifiedDate
private LocalDateTime gmtUpdate;
}
@@ -24,8 +24,7 @@ import javax.persistence.AttributeConverter;
import java.util.Map;
/**
* json 互转map对象字段为数据String字段
*
* json map converter
*
*/
public class JsonMapAttributeConverter implements AttributeConverter<Map<String, String>, String> {
@@ -29,15 +29,14 @@ import static io.swagger.v3.oas.annotations.media.Schema.AccessMode.READ_WRITE;
/**
* 告警 对外上报实体类
*
* Alarm Report Content Entity
*
*/
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
@Schema(description = "告警对外上报字段")
@Schema(description = "Alarm Report Content Entity")
public class AlertReport {
@Schema(title = "Alert record saas index ID")
@@ -24,30 +24,26 @@ import lombok.Data;
import lombok.NoArgsConstructor;
/**
* 监控指标组指标字段
*
* monitoring metrics field
*
*/
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
@Schema(description = "监控指标组指标字段")
@Schema(description = "Monitoring Metrics Field")
public class Field {
@Schema(title = "指标采集字符名称")
@Schema(title = "Metrics Field Name")
private String name;
@Schema(title = "字段类型0-number数字 1-string字符串")
@Schema(title = "Field Type0-number 1-string")
private Byte type;
@Schema(title = "指标单位")
@Schema(title = "Field Unit")
private String unit;
@Schema(title = "是否是实例字段")
private Boolean instance;
@Schema(title = "是否是标签字段")
@Schema(title = "Whether is a label")
private Boolean label;
}
@@ -34,25 +34,25 @@ import static org.dromara.hertzbeat.common.constants.CommonConstants.SUCCESS_COD
*
*/
@Data
@Schema(description = "公共消息包装")
@Schema(description = "Common message structure")
public class Message<T> {
/**
* message body data
*/
@Schema(description = "响应数据")
@Schema(description = "Response Data")
private T data;
/**
* exception message when error happen or success message
*/
@Schema(title = "携带消息")
@Schema(title = "Other Message")
private String msg;
/**
* response code, not http code
*/
@Schema(title = "携带编码")
@Schema(title = "Response Code")
private byte code = SUCCESS_CODE;
public static <T> Message<T> success() {
@@ -90,4 +90,4 @@ public class Message<T> {
private Message(T data) {
this.data = data;
}
}
}
@@ -26,32 +26,31 @@ import lombok.NoArgsConstructor;
import java.util.List;
/**
* 指标组监控数据
*
* Monitoring Metrics Data
*
*/
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
@Schema(description = "指标组监控数据")
@Schema(description = "Monitoring Metrics Data")
public class MetricsData {
@Schema(title = "监控任务ID")
@Schema(title = "Monitoring Task ID")
private Long id;
@Schema(title = "监控类型")
@Schema(title = "Monitoring Type")
private String app;
@Schema(title = "监控指标组")
private String metric;
@Schema(title = "Monitoring Metrics")
private String metrics;
@Schema(title = "最新采集时间")
@Schema(title = "Latest Collect Time")
private Long time;
@Schema(description = "监控指标字段列表")
@Schema(description = "Monitoring Metrics fields")
private List<Field> fields;
@Schema(description = "监控指标列表值集合")
@Schema(description = "Monitoring Metrics DataRow")
private List<ValueRow> valueRows;
}
@@ -27,29 +27,28 @@ import java.util.List;
import java.util.Map;
/**
* 历史单指标数据
*
* Metric History Range Query Data
*
*/
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
@Schema(description = "历史单指标数据")
@Schema(description = "Metric History Range Query Data")
public class MetricsHistoryData {
@Schema(title = "监控任务ID")
@Schema(title = "Monitoring Task ID")
private Long id;
@Schema(title = "监控类型")
@Schema(title = "Monitoring Type")
private String app;
@Schema(title = "监控指标组")
private String metric;
@Schema(title = "Monitoring Metrics")
private String metrics;
@Schema(title = "监控指标")
@Schema(title = "Monitoring Metrics Field")
private Field field;
@Schema(description = "监控指标历史值 instance<==>values")
@Schema(description = "Monitoring Range Query Data tags<==>values")
private Map<String, List<Value>> values;
}
@@ -17,7 +17,6 @@ import java.io.IOException;
import java.util.List;
/**
*
*
*/
@Data

Some files were not shown because too many files have changed in this diff Show More