装置依赖包
yum -y install gcc libffi-devel python-devel openssl-devel4 ca-certificates openssl-dev openssl python2-dev python2 py2-pip py2-yaml libffi-dev gcc musl-dev wget
yum -y install wget openssl openssl-devel gcc gcc-c++
下载 Python3.6.9
wget https://www.python.org/ftp/python/3.6.9/Python-3.6.9.tgz
解压，编译装置
tar xf Python-3.6.9.tgz
cd Python-3.6.9./configure --prefix=/usr/local/python --with-openssl
make && make install
创立软链接，降级 pip
ln -s /usr/local/python/bin/python3 /usr/bin/python3
ln -s /usr/local/python/bin/pip3 /usr/bin/pip3
pip install --upgrade pip
验证
python -V下载安装 elastalert
git clone https://github.com/Yelp/elastalert.git
cd elastalert
pip3 install "elasticsearch<7,>6"
pip3 install -r requirements.txt
python3 setup.py install装置胜利后能够看到四个命令
ll /usr/local/python/bin/elastalert*
/usr/local/python/bin/elastalert
/usr/local/python/bin/elastalert-create-index
/usr/local/python/bin/elastalert-rule-from-kibana
/usr/local/python/bin/elastalert-test-rule
软连贯到 /usr/bin 下, 方便使用
ln -s /usr/local/python/bin/elastalert* /usr/bin
应用
官网文档：https://elastalert.readthedocs.io
规定文档：https://elastalert.readthedocs.io/en/latest/ruletypes.html
依据模板来写 config.yaml
cp config.yaml.example config.yaml
vim config.yaml
配置文件模板
# 用来加载 rule 的目录，默认是 example_rules
rules_folder: rules
# 用来设置定时向 elasticsearch 发送申请，也就是告警执行的频率
run_every:
minutes: 1
# 用来设置申请里工夫字段的范畴
buffer_time:
minutes: 15
# elasticsearch 的 host 地址
es_host: 10.3.0.41
# elasticsearch 的端口
es_port: 9200
# elastalert 产生的日志在 elasticsearch 中的创立的索引
writeback_index: elastalert_status
# 失败重试的工夫限度
alert_time_limit:
#days: 2
minutes: 15
创立告警索引
执行 elastalert-create-index 命令在 ES 创立索引，这不是必须的步骤，然而强烈建议创立。因为对于审计和测试很有用，并且重启 ES 不影响计数和发送 alert.
$ elastalert-create-index
/usr/lib/python2.7/site-packages/elastalert-0.1.37-py2.7.egg/elastalert/create_index.py:65: YAMLLoadWarning: calling yaml.load() without Loader=... is deprecated, as the default Loader is unsafe. Please read https://msg.pyyaml.org/load for full details.
data = yaml.load(config_file)
Elastic Version:6
Mapping used for string:{'type': 'keyword'}
New index elastalert_status created
Done!- Rule 配置
它通过将 Elasticsearch 与两种类型的组件（规定类型和警报）联合应用。定期查问 Elasticsearch，并将数据传递到规定类型，该规定类型确定何时找到匹配项。产生匹配时，将为该警报提供一个或多个警报，这些警报将依据匹配采取行动。
这是由一组规定配置的，每个规定定义一个查问，一个规定类型和一组警报。
ElastAlert 蕴含几种具备常见监督范例的规定类型：
“匹配 Y 工夫有 X 个事件的中央”（frequency type）
“当事件发生率减少或缩小时进行匹配”（spike type）
“在 Y 工夫内少于 X 个事件时进行匹配”（flatline type）
“当某个字段与黑名单 / 白名单匹配时匹配”（blacklist 和 whitelist type）
“匹配与给定过滤器匹配的任何事件”（any type）
“当某个字段在一段时间内具备两个不同的值时进行匹配”（change type）
es_host: 10.13.10.43
es_port: 9200
use_ssl: False
name: Nginx.
use_strftine_index: true
index: logstash-nginx*
type: any
aggregation:
seconds: 10
run_every:
minutes: 1
buffer_time:
minutes: 10
filter:
- query:
query_string:
query: "groups: nginx AND response: 404"
alert:
- "email"
email:
- "7*@qq.net"
smtp_host: smtp.qq.com
smtp_port: 25
smtp_auth_file: /data/elastalert-0.1.37/smtp_auth_file.yaml
from_addr: dbj@qq.com
email_reply_to: dbj@qq.com
启动
python3 -m elastalert.elastalert --verbose --rule rules/nginx.yaml
后续还能够优化启动形式，和告警规定rule
[root@elasticsearch-node1 rules]# pwd
/data/elastalert/rules
[root@elasticsearch-node1 rules]# grep -v "#" nginx.yaml
es_host: 10.216.15.212
es_port: 9200
name: nginx rule
type: frequency
index: nginx_test*
num_events: 3
timeframe:
minutes: 1
filter:
- query:
query_string:
query: "message: 谬误 OR Error"
alert_text: "TTTTTTTtest"
alert:
- "post"
http_post_url: "http://127.0.0.1:9000/elk_alarm"
json
{
"agent": {
"hostname": "elasticsearch-node1",
"name": "elasticsearch-node1",
"id": "f2e24232-1287-45ca-a59f-6a56cec2011c",
"ephemeral_id": "3036c138-4c53-408e-8d4e-2e157885f470",
"type": "filebeat",
"version": "7.8.1"
},
"@timestamp": "2022-05-18T06:48:35.401Z",
"ecs": {"version": "1.5.0"},
"log": {
"file": {"path": "/data/nginx/logs/error.log"},
"offset": 42683
},
"host": {"name": "elasticsearch-node1"},
"@version": "1",
"fields": {"logtype": "test_174"},
"message": "2022/05/18 14:48:30 [error] 21074#0: *267 open()'/data/nginx/html/xxxxxx'failed (2: No such file or directory), client: 10.26.5.20, server: ,request:'HEAD /xxxxxx HTTP/1.1',host:'10.26.5.22'","tags": ["_jsonparsefailure"],"_id":"RwXs1YABqWCT5PWn7_Zj","_index":"nginx_test-2022-05-18","_type":"_doc","num_hits": 10,"num_matches": 3
}
flask code
@alert_list.route('/elk_alarm_all', methods=['POST'])
# @alert_list.route('/elk_alarm/', methods=['POST', 'GET'])
def elk_alarm2():
Time = time.strftime("%Y-%m-%d %H:%M:%S")
data_json = json.loads(request.get_data(as_text=True))
print("###json:", data_json, Time)
messages = data_json['message']
type_log = data_json['agent']['type']
field = data_json['fields']['logtype']
index_log = data_json['_index']
topic_list = index_log.split('-')
topic_name = topic_list[0]
print("###messages:", messages, "filed:", field, "index:", index_log, type_log, topic_name)
logstash_list = logstash.query.filter(logstash.logstash_name == topic_name).first()
logstash_dic = logstash_list.single_to_dict()
group_id = logstash_dic['group_id']
group_users_dic = alert_def.elk_select_alarmgroup(group_id)
temp_elk_dic = alert_def.elk_select_temp(topic_name)
print("###group_user:", group_users_dic, "temp:", temp_elk_dic, type(temp_elk_dic))
temp_elk_dic1 = eval(temp_elk_dic)
group_users = group_users_dic['user_name']
webhook_elk = group_users_dic['webhook']
temp_elk = temp_elk_dic1['tmpl_firing']
change_temp = temp_elk.replace('index_log17', index_log).replace('type_log17', type_log).replace('field17',
field).replace('time17', Time).replace('messages17', messages)
#group_user_list = group_users.split('|')
print("###user:", group_users, "webhook:", webhook_elk, "temp:", change_temp)
alert_def.send_elk_alert_high(temp=change_temp, wx_url=webhook_elk, group_str=group_users)## 日志发送音讯的本人写吧
db.session.close()
return str(data_json)