Версия 13:24, 11 августа 2021

Пример

Это дополнение статьи https://noname.com.ua/mediawiki/index.php?title=Logstash

Задача

Задача - отправка логов от тестового приложения в Logstash

Для тестовых устновок - писать максимально подробные логи приложения
- Nginx - настолько подробно насколько это возможно
- Логи которые пишет бекенд (Ruby)
Для Staging/Prod - менее подробные логи (что бы не перегрузить Elasticsearch)

Реализация

Сервер для сбора - Elasticsearch, Logstash, Kibana, Curator
- Elasticsearch - индексация логов и поиск
- Logstash - прием логов и форматирование перед отравкой в Elasticsearch
- Kibana - интерфейс к Elasticsearch
- Curator -удаление старых индексов
На окружениях которые генерируют логи - Filebeat (Filebeat проще и легковеснее - нет нужды ставить везде Logstash)

При реализации НЕ учитывалось (из экономических соображений):

Отказоустойчивость
- Standalone ElasticSerach (1 нода)
- Один экзкмпляр Logstash
Балансировка нагрузки отсутвует
Нет менеджера очередей для логов для сглаживания пиков нагрузки (можно использовать Kafka/RabbitMQ)

Итого есть три типа логов которы нужно собирать (и они в разных форматах)

nginx access.log
nginx error.log
ruby applocation log

Конфигурация источников логов

Конфигурация nginx

Из-за размера вынесено в отдельную заметку: LogstashExample-nginx-config

Результат

tail -1 access.log | jq .

Пример лога (не все поля)

{
  ...
  "nginx_http_user_agent": "python-requests/2.23.0",
  "nginx_request_headers": " connection='keep-alive'  accept='*/*'  accept-encoding='gzip, deflate'  host='login-anastasiia-env.arturhaunt.com'  user-agent='python-requests/2.23.0' ",
  "nginx_time_iso8601": "2021-08-10T03:09:57+00:00",
  "nginx_time_local": "10/Aug/2021:03:09:57 +0000",
  "nginx_response_body": "<html>\r\n<head><title>301 Moved Permanently</title></head>\r\n<body>\r\n<center><h1>301 Moved Permanently</h1></center>\r\n<hr><center>openresty/1.19.3.2</center>\r\n</body>\r\n</html>\r\n",
  ...
}

Конфигурация приложения

Формат логов приложения в рамках задачи не могут быть изменены

Конфигурация доставки логов

Для доставки логов всех типов используется Filebeat

Общая часть конфигурации Filebeat

Это часть конфигурации которая не относится к какому-то определнному логу а общая для всех.

Модули

Оставега конфигурация по-умолчанию.

filebeat.config.modules:
  path: ${path.config}/modules.d/*.yml
  reload.enabled: false

setup.template.settings:
  index.number_of_shards: 1
setup.kibana:

Output

Отправлять логи в Logstash (на beat input с авторизацией по сертификатам)
Подробнее про настройку: https://noname.com.ua/mediawiki/index.php/Logstash

output.logstash:
  hosts: ["elk.arturhaunt.ninja:5400"]
  ssl.certificate_authorities: ["/etc/elk-certs/elk-ssl.crt"]
  ssl.certificate: "/etc/elk-certs/elk-ssl.crt"
  ssl.key: "/etc/elk-certs/elk-ssl.key"

Процессинг по-умолчанию добавляет метаданные хоста-источника

processors:
  - add_host_metadata:
      when.not.contains.tags: forwarded
  - add_cloud_metadata: ~
  - add_docker_metadata: ~
  - add_kubernetes_metadata: ~
logging.level: info
logging.selectors: ["*"]

Пересылка nginx access logs

Предельно простая конфигурация - читать файлы по маске

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/www/deploy/backend/current/log/nginx/*access.log*
  exclude_files: ['\.gz$']
  fields:
    nginx_logs: "true"
    ruby_application_logs: "false"
    type: "nginx_access_log_json"
    source_hostname: "test-env.domain.tld"
  fields_under_root: true

Пересылка nginx error logs

Ничем не отличается от access.log
Разбор лога на отдельные поля происходит на стороне Logstash.

- type: log
  enabled: true
  paths:
    - /var/www/deploy/backend/current/log/nginx/*error.log*
  exclude_files: ['\.gz$']
  fields:
    nginx_logs: "true"
    ruby_application_logs: "false"
    type: "nginx_error_log"
    source_hostname: "some-domain.tld"
  fields_under_root: true

Пересылка ruby application logs

Немного более сложная сложная конфигурация для многострочного лога

  paths:
    - /var/www/deploy/backend/shared/log/staging.log
  exclude_files: ['\.gz$']
  fields:
    nginx_logs: "false"
    ruby_application_logs: "true"
    type: "ruby_application_logs"
    source_hostname: "some-domain.tld"
  fields_under_root: true
  multiline:
    pattern: '[A-Z], \[[0-9]{4}'
    negate: true
    match: after

Пример лога

D, [2021-08-10T11:17:02.270613 #7] DEBUG -- [ContentSelector::FindExternalService]: /usr/src/app/app/services/content_selector/find_external_service.rb:31:in `perform'
^M/usr/src/app/app/services/service.rb:188:in `call!'
^M/usr/src/app/app/services/service.rb:204:in `call'
^M/usr/src/app/app/services/service.rb:110:in `perform'
^M/usr/src/app/lib/activity_organizers/external.rb:32:in `block in vacant_content_and_topic'
^M/usr/src/app/lib/activity_organizers/external.rb:31:in `each'
^M/usr/src/app/lib/activity_organizers/external.rb:31:in `inject'
^M/usr/src/app/lib/activity_organizers/external.rb:31:in `vacant_content_and_topic'
^M/usr/src/app/lib/activity_organizers/external.rb:18:in `perform'

Разделитель(читается : одна большая буква, запятая, пробел, символ "[", 4 цифры):

pattern: '[A-Z], \[[0-9]{4}'

TODO: Скорее всего список A-Z избыточен так как первая буква означает уровень логгирования, для года наверно можно использовать 202[1-9]

Подробнее про разбор многострочных сообщений - FilebeatMultiline
Оригинальная документация - https://www.elastic.co/guide/en/beats/filebeat/current/multiline-examples.html (но не слишком понятная для меня)
TL;DR: использовать negate: true и match: after практически во всех случаях

Logstash

Input

Получение логов на порту 5400 (Это не syslog протокол - отправляет Filebeat, хотя получене логов через сислог тоже возможно)

input {
    beats {
        port => 5400
        ssl => true
        ssl_certificate_authorities => ["/etc/elk-certs/elk-ssl.crt"]
        ssl_certificate => "/etc/elk-certs/elk-ssl.crt"
        ssl_key => "/etc/elk-certs/elk-ssl.key"
        ssl_verify_mode => "force_peer"
    }
}

В настройке собственно ничего нет кроме путей к сертификатам и номера порта.
Все логи доставляются через один Input

Filter

Общая для всех логов часть

filter {
    mutate {
        add_field => { "nginx" => "nginx_object" }
        add_field => { "ruby" => "ruby_object" }
    }
    mutate {
        rename => { "nginx" => "[application_data][nginx][object]" }
        rename => { "ruby"  => "[application_data][ruby][object]" }
    }
   ... 
   пропущена часть специфичная для каждого из типов логов
   ...
}

Построчный разбор

Добавление полей

Добавить в лог 2 новых поля
Этот шаг нужен для того что бы потом добавлять в эти поля соответствующие логи

    mutate {
        add_field => { "nginx" => "nginx_object" }
        add_field => { "ruby" => "ruby_object" }
    }

Переименование полей

Переименовать поля. По какой-то причине создать сразу поля с вложениями не получается

    mutate {
        rename => { "nginx" => "[application_data][nginx][object]" }
        rename => { "ruby"  => "[application_data][ruby][object]" }
    }

В результате получим

{
  "input": {
    "type": "log"
  "application_data": {
    "ruby": {
      "object": "ruby_object"
    },
    "nginx": {
...

Глобально это сделано для того что бы логи от бекенда и от nginx отличались на уровне application_data а поля верхнего уровня совпадали (но я до сих пор не уверен имеет ли это смысл или может быть лучше писать их в разные индексы)

Filter для nginx access logs

Часть фильтра не относящаяся к nginx access log описана в другом разделе

Конфигурация

filter {
    ... 
    пропущнена часть общая для всех логов
    ...

    if ([type] == "nginx_access_log_json") {
        mutate {
            add_field => { "log_level" => "INFO" }
        }

        json {
            source => "message"
            target => "[application_data][nginx]"
        }

        mutate {
            remove_field => [ "message" ]
        }

        json {
            source => "[application_data][nginx][nginx_request_body]"
            target => "[application_data][nginx][nginx_request_body_json]"
            tag_on_failure => "request_body_is_not_json"
        }

        json {
            source => "[application_data][nginx][nginx_response_body]"
            target => "[application_data][nginx][nginx_response_body_json]"
            tag_on_failure => "response_body_is_not_json"
        }

        kv {
            source => "[application_data][nginx][nginx_request_headers]"
            target => "[application_data][nginx][nginx_request_headers]"
        }
        date {
            match => [
                "[application_data][nginx][nginx_time_iso8601]",
                "MMM dd yyyy HH:mm:ss",
                "MMM  d yyyy HH:mm:ss",
                "ISO8601"
            ]
            target => "@application_timestamp"
        }
    }
...

Преобразования специфичные для access.log

Условие надожения фильтров - далее специфичные для access logs фильтры.

    if ([type] == "nginx_access_log_json") {

Добавление поля с уровнем сообщения

Модификация лога - для всех сообщений из access.log добавляем уровень INFO
TODO(?) Выставлять другой уровень в зависмости от кода ответа?

        mutate {
            add_field => { "log_level" => "INFO" }
        }

Разбор сообщения (т.е. поля "message")

Этот фильтр разбирает строку как JSON.

        json {
            source => "message"
            target => "[application_data][nginx]"
        }

например если Filebeat пришлет строку лога (экранированный JSON)

'{ \"key\": \"value\"} '

то без модификации в логе будет

{
"message": '{ \"key\": \"value\"} '
}

т.е. в поле message сообщение попадет "как есть"
С фильтром же результат будет

{
"application_data": {
  "nginx": {
     "key": "value"
  }
},
"message": '{ \"key\": \"value\"} '
}

"application_data" --> "nginx" описаны в target => "[application_data][nginx]"
Исходный message никуда не девается и удаляить его если не нужен нужно отдельно

Удаление лишнего поля

Разобраное на предыдущем шаге сообшение больше не нужно.
Важно: тут не рассматривается случай еслли вместо JSON в логе прийдет что-то другое, в таком случае это сообшение будет утрачено. Но так-как формат сообщения контролируется со стороны конфигурации nginx то это не проблема.

        mutate {
            remove_field => [ "message" ]
        }

Разбор поля [application_data][nginx][nginx_request_body]

        json {
            source => "[application_data][nginx][nginx_request_body]"
            target => "[application_data][nginx][nginx_request_body_json]"
            tag_on_failure => "request_body_is_not_json"
        }

Тут в случае успеха (т.е. если тело запроса это JSON) - разобранный результат сохранить в поле nginx_request_body_json
Исходное поле не удаляется (на тот случай если в запросе был Не JSON), при ошибке разбора добавляется тег request_body_is_not_json

Разбор поля [application_data][nginx][nginx_response_body_json]

Аналогично - но разбор ответа котрый мог быть JSON или нет

        json {
            source => "[application_data][nginx][nginx_response_body]"
            target => "[application_data][nginx][nginx_response_body_json]"
            tag_on_failure => "response_body_is_not_json"
        }

Разбор поля с заголовками

Поля с заголовками пишуться LUA-кодом:

 rowtext = string.format(" %s='%s' ", k, v)

Это сделано специально для простоты разбора. До разбора заголовки выглядят так:

"nginx_request_headers": " host='domain.tld'  accept-language='en-gb'  accept-encoding='gzip, deflate, br'  connection='keep-alive'  content-type='application/json'  user-agent='Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.1.2 Safari/605.1.15'  authorization='Basic ...='  kbn-system-request='true' ...

Фильтр разбирает такие строки используя разделитель"=" по-умолчанию

        kv {
            source => "[application_data][nginx][nginx_request_headers]"
            target => "[application_data][nginx][nginx_request_headers]"
        }

Этот фильтр берет строку из поля nginx_request_headers и складывает результат работра туде же:

      "nginx_request_headers": {
        "content-type": "application/x-www-form-urlencoded",
        "content-length": "7",
        "host": "...",
        "user-agent": "curl/7.68.0",
        "accept": "*/*",
        "authorization": "Basic ...="

TODO: тут можно подумать как разбирать тело запроса в зависимости от content-type

Разбор поля с датой

Для того что бы использовать время генерации лога, а не время поступления на логстеш.
Настройка максимально проста - поле из которого брать и как "пробовать разобрать".

        date {
            match => [
                "[application_data][nginx][nginx_time_iso8601]",
                "MMM dd yyyy HH:mm:ss",
                "MMM  d yyyy HH:mm:ss",
                "ISO8601"
            ]
            target => "@application_timestamp"
        }

Filter для nginx error logs

Конфигурация

    if ([type] == "nginx_error_log") {
        grok {
            match => [ "message" , "(?<nginx_time>%{YEAR}/%{MONTHNUM}/%{MONTHDAY} %{TIME}) \[%{LOGLEVEL:log_level}\] %{GREEDYDATA:[application_data][nginx][message]}" ]
        }
        mutate {
            remove_field => [ "message" ]
        }
        date {
            match => [
                "[nginx_time]",
                "yyyy/dd/mm HH:mm:ss"
            ]
            target => "@application_timestamp"
        }
        mutate {
            remove_field => [ "nginx_time" ]
        }

    }

Построчный разбор

Фильтр что бы применять

    if ([type] == "nginx_error_log") {

Разбор с помощью grok

        grok {
            match => [ "message" , "(?<nginx_time>%{YEAR}/%{MONTHNUM}/%{MONTHDAY} %{TIME}) \[%{LOGLEVEL:log_level}\] %{GREEDYDATA:[application_data][nginx][message]}" ]
        }

Входные данные выглядят так:

2021/07/06 08:16:18 [notice] 1#1: start worker process 22

Другими словами эьл

Дата и время
Уровень сообщения
Текст сообщения

Конструкция

(?<nginx_time>%{YEAR}/%{MONTHNUM}/%{MONTHDAY} %{TIME})

означает что в результирующее поле nginx_time> записать то что попадет под выражение %{YEAR}/%{MONTHNUM}/%{MONTHDAY} %{TIME}
Далее - стандартная конструкция grok вида %{имя готового выражения:имя поля куда записать}

LOGLEVEL --> log_level
GREEDYDATA (остаток сообщения) --> [application_data][nginx][message] (вложенное поле)

Дебаг GROK: https://grokdebug.herokuapp.com

Удаление оригинальных полей

        mutate {
            remove_field => [ "message" ]
        }

Разбор даты

        date {
            match => [
                "[nginx_time]",
                "yyyy/dd/mm HH:mm:ss"
            ]
            target => "@application_timestamp"
        }

Удаление ненужного поля

        mutate {
            remove_field => [ "nginx_time" ]
        }

Filter для ruby application logs

Конфигурация

    if ([ruby_application_logs] == "true") {
        mutate {
            id => "[Meaningful label for your project, not repeated in any other config files] remove ANSI color codes"
            gsub => ["message", "\x1B\[([0-9]{1,2}(;[0-9]{1,2})?)?[m|K]", ""]
        }

        grok {
            match => [ "message" , "%{WORD:[application_data][ruby][log_level_short]}, \[%{TIMESTAMP_ISO8601:[application_data][ruby][timestamp]} \#%{NUMBER}\] %{WORD:log_level} -- %{GREEDYDATA:[application_data][ruby][log_message]}" ]
        }

        mutate {
            remove_field => [ "message" ]
        }

        date {
            match => [
                "[application_data][ruby][timestamp]",
                "MMM dd yyyy HH:mm:ss",
                "MMM  d yyyy HH:mm:ss",
                "ISO8601"
            ]
            target => "@application_timestamp"
        }

    }

    mutate {
        uppercase => [ "log_level" ]
    }

}

Построчный разбор

Фильтрация

    if ([ruby_application_logs] == "true") {

Удаление цветных кодов

        mutate {
            id => "[Meaningful label for your project, not repeated in any other config files] remove ANSI color codes"
            gsub => ["message", "\x1B\[([0-9]{1,2}(;[0-9]{1,2})?)?[m|K]", ""]
        }

Разбор с помощью grok

        grok {
            match => [ "message" , "%{WORD:[application_data][ruby][log_level_short]}, \[%{TIMESTAMP_ISO8601:[application_data][ruby][timestamp]} \#%{NUMBER}\] %{WORD:log_level} -- %{GREEDYDATA:[application_data][ruby][log_message]}" ]
        }

Поле message разбирается на поля

[application_data][ruby][log_level_short] - короткое однобуквенное обозначение уровня лога (D - Debug, E - Error и т.д.)
[application_data][ruby][timestamp] - время
log_level - уровень лога, словом
[application_data][ruby][log_message] - само сообщение до колнца строки

Дебаг GROK: https://grokdebug.herokuapp.com (что б не забыть)

Удаление ненужного поля

        mutate {
            remove_field => [ "message" ]
        }

Получение даты

        date {
            match => [
                "[application_data][ruby][timestamp]",
                "MMM dd yyyy HH:mm:ss",
                "MMM  d yyyy HH:mm:ss",
                "ISO8601"
            ]
            target => "@application_timestamp"
        }

Преобразование к заглавным буквам

нужно так как приложения могут писать уровень по-разному - 'Error', 'error', 'ERROR'

    mutate {
        uppercase => [ "log_level" ]
    }

TODO Замена 'Err' -> Error и т.п. если найдется

Output

Общая для всех логов часть

Общий получатель логов - Elasticsearch

output {
    elasticsearch {
        hosts => ["localhost:9200"]
        index => "application-logs-%{[host][name]}-%{+YYYY.MM.dd}"
        document_type => "nginx_logs"
    }
    ...
    пропущены отладочные получатели
    ...
}

Отладка

Для того что б удобно смотреть на результат работы фильтров и преобразований самый простой способ - писать логи в файл.

Для отладки для nginx access logs

#    if ([type] == "nginx_access_log_json") {
#        file {
#            flush_interval => 5
#                gzip => false
#                path => "/var/log/logstash/nginx-%{[host][name]}-%{+YYYY.MM.dd}.log"
#        }
#    }

Для отладки для nginx error logs

#    if ([type] == "nginx_error_log") {
#        file {
#            flush_interval => 5
#                gzip => false
#                path => "/var/log/logstash/nginx-errors-%{[host][name]}-%{+YYYY.MM.dd}.log"
#        }
#    }

Для отладки для ruby application logs

#
#    if ([ruby_application_logs] == "true") {
#        file {
#            flush_interval => 5
#                gzip => false
#                path => "/var/log/logstash/ruby-%{[host][name]}-%{+YYYY.MM.dd}.log"
#        }
#    }

Elasticsearch-Curator

Ссылки

https://pawelurbanek.com/elk-nginx-logs-setup

@@ Строка 641: / Строка 641: @@
 #        }
 #    }
-<PRE>
+</PRE>
 =Elasticsearch-Curator=

LogstashExample1: различия между версиями

Версия 13:24, 11 августа 2021

Пример

Задача

Реализация

Конфигурация источников логов

Конфигурация nginx

Результат

Конфигурация приложения

Конфигурация доставки логов

Общая часть конфигурации Filebeat

Модули

Output

Пересылка nginx access logs

Пересылка nginx error logs

Пересылка ruby application logs

Logstash

Input

Filter

Общая для всех логов часть

Построчный разбор

Добавление полей

Переименование полей

Filter для nginx access logs

Конфигурация

Преобразования специфичные для access.log

Добавление поля с уровнем сообщения

Разбор сообщения (т.е. поля "message")

Удаление лишнего поля

Разбор поля [application_data][nginx][nginx_request_body]

Разбор поля [application_data][nginx][nginx_response_body_json]

Разбор поля с заголовками

Разбор поля с датой

Filter для nginx error logs

Конфигурация

Построчный разбор

Фильтр что бы применять

Разбор с помощью grok

Удаление оригинальных полей

Разбор даты

Удаление ненужного поля

Filter для ruby application logs

Конфигурация

Построчный разбор

Фильтрация

Удаление цветных кодов

Разбор с помощью grok

Удаление ненужного поля

Получение даты

Преобразование к заглавным буквам

Output

Общая для всех логов часть

Отладка

Для отладки для nginx access logs

Для отладки для nginx error logs

Для отладки для ruby application logs

Elasticsearch-Curator

Ссылки

Навигация

Поиск