1

我有来自各种来源的日志，日志的格式是

[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 堆栈进行日志聚合的最佳实践是什么？

4

1 回答 1

0

如果您想坚持 Elastic 堆栈的组件，可以读取、解析和持久化日志，如下所示：

  1. Filbeat：读取事件（日志的每个逻辑行）并将其推送到 Logstash
  2. Logstash：解析日志以根据您的要求将字符串分解为有意义的字段。可以使用 GROK 过滤器解析字符串。这比构建自定义解析器更可取。解析后的信息被发送到 Elasticsearch 以进行持久化和索引，最好基于时间戳。
  3. Kibana：使用单一搜索或聚合可视化解析的信息。
于 2018-11-29T15:44:16.203 回答