Предположим, что мы используем для сбора логов стек ELK (Elasticsearch-Logstash-Kibana).

Логи приложений можно собирать как в один индекс, так и отдельные индексы для каждого из приложений.

В этом блоге рассмотрим маршрутизацию записей логов в разные индексы.

Для доставки файлов в Logstash будем использовать Filebeat

Рассмотрим пример маркирования логов системной утилитой auditd маркером "auditd: true" и маркирование логов PostgreSQL маркером "postgres: true". В примере Filebeat отправляет записи логов в Logstash по порту 5044.

Пример конфигурации filebeat.yml:

- input_type: log
  paths:
    - /var/log/audit/*.log
  fields:
    auditd: true
- input_type: log
  paths:
    - /var/lib/pgsql/12/data/log/postgresql-*.log
  fields:
    postgres: true
output.logstash:
  hosts: ["localhost:5044"]

Теперь, непосредственно, рассмотрим маршрутизацию логов в Logstash в logstash.conf.

input {
  beats {
    port => 5044
  }
}
output {
        if [fields][auditd] {
                elasticsearch {
                        hosts => ["localhost:9200"]
                        index => "auditd-%{+YYYY.MM.dd}"
                }
        } else if [fields][postgres] {
                elasticsearch {
                        hosts => ["localhost:9200"]
                        index => "pgaudit-%{+YYYY.MM.dd}"
                }
        } else {
                elasticsearch {
                        hosts => ["localhost:9200"]
                        index => "unknown-%{+YYYY.MM.dd}"
                }
        }
	stdout { codec => rubydebug }
}

В примере описанном выше Logstash ведет прием записей логов по порту 5044 и делает маршрутизацию в зависимости от флагов fields.auditd и fields.postgres. Если не получается распределить сообщения в требуемый индекс, то отправляем его в индекс под названием unknown.

Индексы в Elasticsearch записываются по дневным партициям: %{+YYYY.MM.dd}.