ElastAlert核心逻辑流程及源码解析

set_starttime：设置开始时间1. scan_entire_timeframe为设置时：查询ES获取最后执行时间2. use_count_query和use_terms_query未设置时：  2.1 scan_entire_timeframe开启时：endtime - rule['timeframe']  2.2 else：endtime - rule.get('buffer_time')3. else：  3.1 scan_entire_timeframe开启时：endtime - rule['timeframe']  3.1 else：endtime - self.run_every

init_rule：从存在的规则中复制出非状态配置到一个新规则1. 判断不是新规则，scheduler.remove_job(rule['name'])2. 判断ES5、增强过滤、规则参数设置3. scheduler.add_job(handle_rule_execution)

handle_rule_execution：规则处理1. 获得延迟查询的时间：endtime = ts_now() -rule['query_delay']2. 调用run_rule：运行规则3. 调用remove_old_events：移除rule['processed_hits']中超过buffer_time的事件4. 调用reset_rule_schedule：修改job的下一次执行时间

handle_config_change：通过规则配置文件修改时间，重新加载规则

5

handle_pending_alerts->send_pending_alerts：发送待定的警报1. find_recent_pending_alerts(rule['alert_time_limit'])，查询回写索引，重试窗口期中待警报事件2. scheduler.add_job(handle_config_change)：3. sleep_for：暂停到下一次执行时间

2.1

start：定期检查并运行规则1. scheduler.add_job(send_pending_alerts)：2. scheduler.add_job(handle_config_change)：3. sleep_for：暂停到下一次执行时间

process_hits()：根据规则配置更新_source字段值1. 时间戳转换2. 将重要字段添加到hit['_source'][field]3. 将compound_query_key字段值连接添加到hit['_source'][rule['query_key']]字段中4. 将compound_aggregation_key字段值连接添加到hit['_source'][rule['aggregation_key']]

2.4

2

get_segment_size：可以是buffer_size叠加，也可以是run_every严格分开1. use_count_query&use_terms_query&aggregation_query_element未设置时：return buffer_time2. aggregation_query_element设置时：  2.1 use_run_every_query_size设置时：return run_every  2.2 else：return buffer_time3. else：return run_every

3

handle_pending_alerts->send_pending_alerts：发送延迟警报1. top_count_keys存在，span style=\"font-size: inherit;\

2.3

2.2

__init__1. 初始化参数2. 加载所有规则

1

main1. 创建对象ElastAlerter2. 调用start()