Logstash: конфигурируем сбор логов

В данной статье рассмотрим варианты конфигураций логстеша(опции filter{}).
Для начала сделаем авто рестарт конфигурации, чтобы постоянно не перезапускать службу логстеша. Добавляем к параметрам запуска параметр config.reload.automatic
ExecStart=/usr/share/logstash/bin/logstash  ...  "--config.reload.automatic"
Далее => systemctl daemon-reload и перезапуск самого логстеша.

filter{ ...}
#пример форматирования даты-времени
 if "json" in [tags] { #находим в тегах json и применяем данный паттерн
    date {
      match => ["[json][@timestamp]", "yyyy-MM-dd HH:mm:ss"]
      timezone => "Europe/Moscow" #применяем временные настройки
      target => "[json][@timestamp]"
    }
  }

#пример поиска текста по полю [json][message], который не равен тексту ERROR
if [json][message] !~ "ERROR"{
        mutate {
            remove_field => ["input"]} #удаляем поле из вывода
           rename => { "FName" => "Firstname"} #переименование

   gsub => [ "EmailId", "\.", "_" ] #замена . в поле EmailId на _

           strip => ["Firstname", "Lastname"]

           uppercase => [ "Gender" ]

}
mutate позволяет выполнять различные изменения данных в полях. Вы можете переименовать, конвертировать, извлекать и модифицировать поля в событиях.
Параметр gsub используется для сопоставления регулярного выражения и значения поля, а также для замены всех соответствий замещающей строкой.
Параметр strip используется для извлечения пробелов.
Параметр uppercase предназначен для преобразования строки в верхний регистр.

Фильтр Grok

Это мощный и популярный плагин для преобразования неструктурированных данных в структурированные, для упрощения запросов и фильтрации этих данных. В общих чертах Grok позволяет сопоставить строку и шаблон (который основан на регулярном выражении) и отобразить определенные части строки на свои поля. Общий синтаксис шаблона Grok выглядит следующим образом:

%{PATTERN:FIELDNAME}

PATTERN — имя шаблона, который соответствует тексту. FIELDNAME — идентификатор для соответствующего фрагмента текста.

#пример для Apache, двойной грок, т.к. не всегда апач заполняет все поля. В этом случае вывод не формируется. В примере указано, что если поле msg заполнено, то его нужно отфильтровать.
if "apache_access" in [tags] {
    grok {
                match => { "message" => "%{IPORHOST:[remote_ip]} %{IPORHOST:[next_ip]} %{IPORHOST:[remote_ip1]} - \[%{HTTPDATE:[timestamp]}\] \"%{WORD:[method]} %{DATA:[url]} HTTP/%{NUMBER:[http_version]}\" %{NUMBER:response_code:int} %{NUMBER:bytes:int} \"-\"\s*%{GREEDYDATA:msg}"}
                remove_field => "message"
    }
    grok {
                match => [ "msg", "\s*\*\*%{NUMBER:referrer:int}/%{NUMBER:referrer2:int}\*\*\s*%{GREEDYDATA:user}" ]
    }
    mutate {
        remove_field => ["input", "ecs", "@version", "agent", "next_ip", "http_version", "remote_ip1"]
    }
  }

Вот так выглядит строка из лога, которую нужно пройти двойным гроком.
Jan 31, 2020 @ 11:09:58.304     -     111.222.233.119 10.218.188.117 123.222.2.119 - [31/Jan/2020:11:09:58 +0300] "POST /url/somemethod HTTP/1.1" 200 457 "-" **0/54046** user_some

С помощбю Kibana вы можете отфильтровать свой лог. Указываете формат, в котором поступают данные и далее с помощью фильтров настраиваете формат лога.

Успехов!