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

Материал из noname.com.ua
Перейти к навигацииПерейти к поиску
Строка 560: Строка 560:
 
* log_level - уровень лога, словом
  
* log_level - уровень лога, словом
 
* [application_data][ruby][log_message] - само сообщение до колнца строки
  
* [application_data][ruby][log_message] - само сообщение до колнца строки
  +
  +
* Дебаг GROK: https://grokdebug.herokuapp.com (что б не забыть)
   
 
=====Удаление ненужного поля=====
  
=====Удаление ненужного поля=====

Версия 13:03, 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

Результат

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]

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] - само сообщение до колнца строки
Удаление ненужного поля
        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
Источник — https://noname.com.ua/mediawiki/index.php?title=LogstashExample1&oldid=9081