fluentd: объединение журналов из нескольких ресурсов с использованием Elastic search FluentD и Kibana

Question

У меня есть журналы, поступающие из разных источников, и формат журналов

[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?

Answer 1

Если вы хотите придерживаться компонентов стека Elastic, журналы можно читать, анализировать и сохранять, как показано ниже:

1. Filbeat: читает события (каждую логическую строку журналов) и отправляет их в Logstash
2. Logstash: разобрать журналы, чтобы разбить строки на значимые поля в соответствии с вашими требованиями. Строки могут быть проанализированы с помощью фильтров GROK. Это предпочтительнее, чем создание пользовательских парсеров. Проанализированная информация отправляется в Elasticsearch для сохранения и индексации предпочтительно на основе метки времени.
3. Кибана: визуализировать разобранную информацию с помощью одного поиска или агрегации.