elasticsearch - fluentd :使用弹性搜索 FluentD 和 Kibana 从多个资源聚合日志
问题描述
我有来自各种来源的日志,日志的格式是
[2018-11-20 11:27:41,187] {base_task.py:98} INFO - Subtask: [2018-11-20 11:27:41,186] {child_task.py:355} INFO - Inside poll job status
[2018-11-20 11:27:41,187] {base_task.py:98} INFO - Subtask: [2018-11-20 11:27:41,186] {child_task.py:357} DEBUG - Poll time out has been set to: 6 hr(s)
[2018-11-20 11:27:41,188] {base_task.py:98} INFO - Subtask: [2018-11-20 11:27:41,186] {child_task.py:369} DEBUG - Batch_id of the running job is = 123456
[2018-11-20 11:27:41,188] {base_task.py:98} INFO - Subtask: [2018-11-20 11:27:41,186] {child_task.py:377} DEBUG - Getting cluster ID for the cluster:
我想将这些日志推送到索引为 batch_id 的弹性搜索中,我该如何实现?问题是我在某些行中有 batch_id,而不是在所有行中。我已经编写了自定义解析器来将日志转换为 JSON
td-agent.conf 是
<source>
@type tail
path /tmp/logs/airflow.logs
pos_file /tmp/logs/airflow1.pos
format /^\[(?<logtime>[^\]]*)\] \{(?<parent_script>[^ ]*)\} (?<parent_script_log_level>[^ ]*) - (?<subtask_name>[^ ]*): \[(?<subtask_log_time>[^\]]*)\] \{(?<script_name>[^ ]*)\} (?<script_log_info>[^ ]*) - (?<message>[^*]*)/
time_key logtime
tag airflow_123
read_from_head true
include_tag_key true
tag_key event_tag
@log_level debug
</source>
<match airflow_123>
@type copy
<store>
@type stdout
</store>
<store>
@type elasticsearch
host es_host
port es_port
index_name fluentd.${tag}.%Y%m%d
<buffer tag, time>
timekey 1h # chunks per hours ("3600" also available)
</buffer>
type_name log
with_transporter_log true
@log_level debug
</store>
</match>
此外,使用 EFK 堆栈进行日志聚合的最佳实践是什么?
解决方案
如果您想坚持 Elastic 堆栈的组件,可以读取、解析和持久化日志,如下所示:
- Filbeat:读取事件(日志的每个逻辑行)并将其推送到 Logstash
- Logstash:解析日志以根据您的要求将字符串分解为有意义的字段。可以使用 GROK 过滤器解析字符串。这比构建自定义解析器更可取。解析后的信息被发送到 Elasticsearch 以进行持久化和索引,最好基于时间戳。
- Kibana:使用单一搜索或聚合可视化解析的信息。