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

Материал из noname.com.ua
Перейти к навигацииПерейти к поиску
Строка 382: Строка 382:
 
"message": '{ \"key\": \"value\"} '
  
"message": '{ \"key\": \"value\"} '
 
}
  
}
−
<PRE>
 +
</PRE>
 
т.е. в поле message сообщение попадет "как есть"
  
т.е. в поле message сообщение попадет "как есть"
 
<BR>
  
<BR>
Строка 396: Строка 396:
 
}
  
}
 
</PRE>
  
</PRE>
  +
"application_data" --> "nginx" описаны в target => "[application_data][nginx]"
   
 
=====Добавление полей=====
  
=====Добавление полей=====

Версия 12:51, 10 августа 2021

Пример

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

Задача

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

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

Реализация

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


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

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

nginx

Nginx умеет писать логи в Json (для максимально подробного логгирования требуется поддержка Lua)

Конфигурация (только значимые части)

   # Log in JSON Format
    log_format nginxlog_json escape=json
    '{ '
    '"nginx_http_user_agent": "$http_user_agent",'
    '"nginx_ancient_browser": "$ancient_browser",'
    '"nginx_body_bytes_sent": "$body_bytes_sent",'
    '"nginx_bytes_sent": "$bytes_sent",'
    '"nginx_connection": "$connection",'
    '"nginx_connection_requests": "$connection_requests",'
    '"nginx_connections_active": "$connections_active",'
    '"nginx_connections_reading": "$connections_reading",'
    '"nginx_connections_waiting": "$connections_waiting",'
    '"nginx_connections_writing": "$connections_writing",'
    '"nginx_content_length": "$content_length",'
    '"nginx_content_type": "$content_type",'
    '"nginx_cookie_": "$cookie_",'
    '"nginx_document_root": "$document_root",'
    '"nginx_document_uri": "$document_uri",'
    '"nginx_fastcgi_path_info": "$fastcgi_path_info",'
    '"nginx_fastcgi_script_name": "$fastcgi_script_name",'
    '"nginx_host": "$host",'
    '"nginx_hostname": "$hostname",'
    '"nginx_https": "$https",'
    '"nginx_invalid_referer": "$invalid_referer",'
    '"nginx_is_args": "$is_args",'
    '"nginx_limit_conn_status": "$limit_conn_status",'
    '"nginx_limit_rate": "$limit_rate",'
    '"nginx_limit_req_status": "$limit_req_status",'
    '"nginx_modern_browser": "$modern_browser",'
    '"nginx_msec": "$msec",'
    '"nginx_msie": "$msie",'
    '"nginx_nginx_version": "$nginx_version",'
    '"nginx_proxy_add_x_forwarded_for": "$proxy_add_x_forwarded_for",'
    '"nginx_proxy_host": "$proxy_host",'
    '"nginx_proxy_port": "$proxy_port",'
    '"nginx_proxy_protocol_addr": "$proxy_protocol_addr",'
    '"nginx_proxy_protocol_port": "$proxy_protocol_port",'
    '"nginx_proxy_protocol_server_addr": "$proxy_protocol_server_addr",'
    '"nginx_proxy_protocol_server_port": "$proxy_protocol_server_port",'
    '"nginx_query_string": "$query_string",'
    '"nginx_realip_remote_addr": "$realip_remote_addr",'
    '"nginx_realip_remote_port": "$realip_remote_port",'
    '"nginx_remote_addr": "$remote_addr",'
    '"nginx_remote_port": "$remote_port",'
    '"nginx_remote_user": "$remote_user",'
    '"nginx_request": "$request",'
    '"nginx_request_headers": "$request_headers",'
    '"nginx_request_body": "$request_body",'
    '"nginx_request_id": "$request_id",'
    '"nginx_request_length": "$request_length",'
    '"nginx_request_method": "$request_method",'
    '"nginx_request_time": "$request_time",'
    '"nginx_request_uri": "$request_uri",'
    '"nginx_scheme": "$scheme",'
    '"nginx_server_addr": "$server_addr",'
    '"nginx_server_name": "$server_name",'
    '"nginx_server_port": "$server_port",'
    '"nginx_server_port": "$server_port",'
    '"nginx_server_protocol": "$server_protocol",'
    '"nginx_ssl_cipher": "$ssl_cipher",'
    '"nginx_ssl_ciphers": "$ssl_ciphers",'
    '"nginx_ssl_client_cert": "$ssl_client_cert",'
    '"nginx_ssl_client_escaped_cert": "$ssl_client_escaped_cert",'
    '"nginx_ssl_client_fingerprint": "$ssl_client_fingerprint",'
    '"nginx_ssl_client_i_dn": "$ssl_client_i_dn",'
    '"nginx_ssl_client_raw_cert": "$ssl_client_raw_cert",'
    '"nginx_ssl_client_s_dn": "$ssl_client_s_dn",'
    '"nginx_ssl_client_serial": "$ssl_client_serial",'
    '"nginx_ssl_client_v_end": "$ssl_client_v_end",'
    '"nginx_ssl_client_v_remain": "$ssl_client_v_remain",'
    '"nginx_ssl_client_v_start": "$ssl_client_v_start",'
    '"nginx_ssl_client_verify": "$ssl_client_verify",'
    '"nginx_ssl_early_data": "$ssl_early_data",'
    '"nginx_ssl_protocol": "$ssl_protocol",'
    '"nginx_ssl_server_name": "$ssl_server_name",'
    '"nginx_ssl_session_id": "$ssl_session_id",'
    '"nginx_ssl_session_reused": "$ssl_session_reused",'
    '"nginx_status": "$status",'
    '"nginx_tcpinfo_rtt": "$tcpinfo_rtt",'
    '"nginx_tcpinfo_rttvar": "$tcpinfo_rttvar",'
    '"nginx_tcpinfo_snd_cwnd": "$tcpinfo_snd_cwnd",'
    '"nginx_tcpinfo_rcv_space": "$tcpinfo_rcv_space",'
    '"nginx_time_iso8601": "$time_iso8601",'
    '"nginx_time_local": "$time_local",'
    '"nginx_uid_got": "$uid_got",'
    '"nginx_uid_reset": "$uid_reset",'
    '"nginx_uid_set": "$uid_set",'
    '"nginx_upstream_addr": "$upstream_addr",'
    '"nginx_upstream_bytes_received": "$upstream_bytes_received",'
    '"nginx_upstream_bytes_sent": "$upstream_bytes_sent",'
    '"nginx_upstream_bytes_sent": "$upstream_bytes_sent",'
    '"nginx_upstream_cache_status": "$upstream_cache_status",'
    '"nginx_upstream_connect_time": "$upstream_connect_time",'
    '"nginx_upstream_cookie_": "$upstream_cookie_",'
    '"nginx_upstream_header_time": "$upstream_header_time",'
    '"nginx_upstream_http_": "$upstream_http_",'
    '"nginx_upstream_response_length": "$upstream_response_length",'
    '"nginx_upstream_response_time": "$upstream_response_time",'
    '"nginx_upstream_status": "$upstream_status",'
    '"nginx_uri": "$uri",'
    '"nginx_response_body": "$response_body"'
    '}';

    server {
        listen 443 ssl;
        root /var/www/backend;
        server_name elk.domain.tld;
        access_log /var/log/nginx/elk.domain.tld-access.log.ssl nginxlog_json;
        error_log /var/log/nginx/elk.domain.tld-error.log.ssl;
        client_max_body_size 500M;
        keepalive_timeout 0;
        ssl_certificate     ...;
        ssl_certificate_key ...;

        lua_need_request_body on;

        set $response_body "";
        body_filter_by_lua '
            local response_body = string.sub(ngx.arg[1], 1, 1000)
            ngx.ctx.buffered = (ngx.ctx.buffered or "") .. response_body
            if ngx.arg[2] then
                ngx.var.response_body = ngx.ctx.buffered
            end
        ';

        set_by_lua_block $request_headers{
            local h = ngx.req.get_headers()
            local request_headers_all = ""
            for k, v in pairs(h) do
                local rowtext = ""
                rowtext = string.format(" %s='%s' ", k, v)
                request_headers_all = request_headers_all .. rowtext

            end
            return request_headers_all
        }
...

Результат

tail -1 access.log | jq .

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

{
  ...
  "nginx_http_user_agent": "python-requests/2.23.0",
  "nginx_ancient_browser": "1",
  "nginx_body_bytes_sent": "175",
  "nginx_document_root": "/usr/local/openresty/nginx/html",
  "nginx_document_uri": "/favicon.ico",
  "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 для логов Nginx

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

Модули (по-умолчанию) 

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

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

Отправлять логи в 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: ["*"]

Logstash для nginx

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"
    }
}

В настройке собственно ничего нет кроме путей к сертификатам и номера порта.

Filter

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

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

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]" }
    }

    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

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

Добавить в лог 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 а поля верхнего уровня совпадали (но я до сих пор не уверен имеет ли это смысл или может быть лучше писать их в разные индексы)

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


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

    if ([type] == "nginx_access_log_json") {
Добавление поля с уровнем сообщения

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

        mutate {
            add_field => { "log_level" => "INFO" }
        }
Разбор сообщения

Этот фильтр разбирает строку как 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]"

Добавление полей
        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"
        }

Output

11

В результате сообщение будет разобрано на поля - но в этом случае в поле host 

  "host": {
    "hostname": "elk.domain.tld",
    "id": "ec2385ad84aeea5eb425460c41a5b866",
    "os": {
      "type": "linux",
      "codename": "focal",
      "name": "Ubuntu",
      "version": "20.04.2 LTS (Focal Fossa)",
      "kernel": "5.4.0-1045-aws",
      "platform": "ubuntu",
      "family": "debian"
    },

будет записан заголовок HOST что не cовсем то что надо - оригинально там содержаться данные о том какой хост отправил логи

Пример с дебагом

Усложняем - добавляем максимальное количесво информации в логи. В том числе тело запроса и тело ответа и все заголовки которые возможно.

Nginx

Пишем все что возможно: https://noname.com.ua/mediawiki/index.php/Nginx_Log_Post 

   # Log in JSON Format
    log_format nginxlog_json escape=json
    '{ '
    '"nginx_http_user_agent": "$http_user_agent",'
    '"nginx_ancient_browser": "$ancient_browser",'
    '"nginx_body_bytes_sent": "$body_bytes_sent",'
    '"nginx_bytes_sent": "$bytes_sent",'
    '"nginx_connection": "$connection",'
    '"nginx_connection_requests": "$connection_requests",'
    '"nginx_connections_active": "$connections_active",'
    '"nginx_connections_reading": "$connections_reading",'
    '"nginx_connections_waiting": "$connections_waiting",'
    '"nginx_connections_writing": "$connections_writing",'
    '"nginx_content_length": "$content_length",'
    '"nginx_content_type": "$content_type",'
    '"nginx_cookie_": "$cookie_",'
    '"nginx_document_root": "$document_root",'
    '"nginx_document_uri": "$document_uri",'
    '"nginx_fastcgi_path_info": "$fastcgi_path_info",'
    '"nginx_fastcgi_script_name": "$fastcgi_script_name",'
    '"nginx_host": "$host",'
    '"nginx_hostname": "$hostname",'
    '"nginx_https": "$https",'
    '"nginx_invalid_referer": "$invalid_referer",'
    '"nginx_is_args": "$is_args",'
    '"nginx_limit_conn_status": "$limit_conn_status",'
    '"nginx_limit_rate": "$limit_rate",'
    '"nginx_limit_req_status": "$limit_req_status",'
    '"nginx_modern_browser": "$modern_browser",'
    '"nginx_msec": "$msec",'
    '"nginx_msie": "$msie",'
    '"nginx_nginx_version": "$nginx_version",'
    '"nginx_proxy_add_x_forwarded_for": "$proxy_add_x_forwarded_for",'
    '"nginx_proxy_host": "$proxy_host",'
    '"nginx_proxy_port": "$proxy_port",'
    '"nginx_proxy_protocol_addr": "$proxy_protocol_addr",'
    '"nginx_proxy_protocol_port": "$proxy_protocol_port",'
    '"nginx_proxy_protocol_server_addr": "$proxy_protocol_server_addr",'
    '"nginx_proxy_protocol_server_port": "$proxy_protocol_server_port",'
    '"nginx_query_string": "$query_string",'
    '"nginx_realip_remote_addr": "$realip_remote_addr",'
    '"nginx_realip_remote_port": "$realip_remote_port",'
    '"nginx_remote_addr": "$remote_addr",'
    '"nginx_remote_port": "$remote_port",'
    '"nginx_remote_user": "$remote_user",'
    '"nginx_request": "$request",'
    '"nginx_request_headers": "$request_headers",'
    '"nginx_request_body": "$request_body",'
    '"nginx_request_id": "$request_id",'
    '"nginx_request_length": "$request_length",'
    '"nginx_request_method": "$request_method",'
    '"nginx_request_time": "$request_time",'
    '"nginx_request_uri": "$request_uri",'
    '"nginx_scheme": "$scheme",'
    '"nginx_server_addr": "$server_addr",'
    '"nginx_server_name": "$server_name",'
    '"nginx_server_port": "$server_port",'
    '"nginx_server_port": "$server_port",'
    '"nginx_server_protocol": "$server_protocol",'
    '"nginx_ssl_cipher": "$ssl_cipher",'
    '"nginx_ssl_ciphers": "$ssl_ciphers",'
    '"nginx_ssl_client_cert": "$ssl_client_cert",'
    '"nginx_ssl_client_escaped_cert": "$ssl_client_escaped_cert",'
    '"nginx_ssl_client_fingerprint": "$ssl_client_fingerprint",'
    '"nginx_ssl_client_i_dn": "$ssl_client_i_dn",'
    '"nginx_ssl_client_raw_cert": "$ssl_client_raw_cert",'
    '"nginx_ssl_client_s_dn": "$ssl_client_s_dn",'
    '"nginx_ssl_client_serial": "$ssl_client_serial",'
    '"nginx_ssl_client_v_end": "$ssl_client_v_end",'
    '"nginx_ssl_client_v_remain": "$ssl_client_v_remain",'
    '"nginx_ssl_client_v_start": "$ssl_client_v_start",'
    '"nginx_ssl_client_verify": "$ssl_client_verify",'
    '"nginx_ssl_early_data": "$ssl_early_data",'
    '"nginx_ssl_protocol": "$ssl_protocol",'
    '"nginx_ssl_server_name": "$ssl_server_name",'
    '"nginx_ssl_session_id": "$ssl_session_id",'
    '"nginx_ssl_session_reused": "$ssl_session_reused",'
    '"nginx_status": "$status",'
    '"nginx_tcpinfo_rtt": "$tcpinfo_rtt",'
    '"nginx_tcpinfo_rttvar": "$tcpinfo_rttvar",'
    '"nginx_tcpinfo_snd_cwnd": "$tcpinfo_snd_cwnd",'
    '"nginx_tcpinfo_rcv_space": "$tcpinfo_rcv_space",'
    '"nginx_time_iso8601": "$time_iso8601",'
    '"nginx_time_local": "$time_local",'
    '"nginx_uid_got": "$uid_got",'
    '"nginx_uid_reset": "$uid_reset",'
    '"nginx_uid_set": "$uid_set",'
    '"nginx_upstream_addr": "$upstream_addr",'
    '"nginx_upstream_bytes_received": "$upstream_bytes_received",'
    '"nginx_upstream_bytes_sent": "$upstream_bytes_sent",'
    '"nginx_upstream_bytes_sent": "$upstream_bytes_sent",'
    '"nginx_upstream_cache_status": "$upstream_cache_status",'
    '"nginx_upstream_connect_time": "$upstream_connect_time",'
    '"nginx_upstream_cookie_": "$upstream_cookie_",'
    '"nginx_upstream_header_time": "$upstream_header_time",'
    '"nginx_upstream_http_": "$upstream_http_",'
    '"nginx_upstream_response_length": "$upstream_response_length",'
    '"nginx_upstream_response_time": "$upstream_response_time",'
    '"nginx_upstream_status": "$upstream_status",'
    '"nginx_uri": "$uri",'
    '"nginx_response_body": "$response_body"'
    '}';

    server {
        listen 443 ssl;
        root /var/www/backend;
        server_name elk.domain.tld;
        access_log /var/log/nginx/elk.domain.tld-access.log.ssl nginxlog_json;
        error_log /var/log/nginx/elk.domain.tld-error.log.ssl;
        client_max_body_size 500M;
        keepalive_timeout 0;
        ssl_certificate     ...;
        ssl_certificate_key ...;

        lua_need_request_body on;

        set $response_body "";
        body_filter_by_lua '
            local response_body = string.sub(ngx.arg[1], 1, 1000)
            ngx.ctx.buffered = (ngx.ctx.buffered or "") .. response_body
            if ngx.arg[2] then
                ngx.var.response_body = ngx.ctx.buffered
            end
        ';

        set_by_lua_block $request_headers{
            local h = ngx.req.get_headers()
            local request_headers_all = ""
            for k, v in pairs(h) do
                local rowtext = ""
                rowtext = string.format(" %s='%s' ", k, v)
                request_headers_all = request_headers_all .. rowtext

            end
            return request_headers_all
        }
...

Logstash

filter {
    json {
        source => "message"
    }
    mutate {
        remove_field => [ "message" ]
    }
    json {
        source => "nginx_request_body"
        target => "nginx_request_body_json"
        tag_on_failure => "request_body_is_not_json"
    }

    json {
        source => "nginx_response_body"
        target => "nginx_response_body_json"
        tag_on_failure => "response_body_is_not_json"
    }

    kv {
        source => "nginx_request_headers"
        target => "nginx_request_headers"
    }
}

Разобрать message как JSON и записать на верхний уровень. Это значит что все поля из сообщения станут полями верхнего уровня (и перепишут если такие поля уже были - по этой причине в логи nginx добавлен префикс nginx_ 

json {
    source => "message"
}

Удалить оригинальное поле 

mutate {
    remove_field => [ "message" ]
}

Разобрать (если это возможно) тело запроса и записать в ключ nginx_request_body_json 

json {
    source => "nginx_request_body"
    target => "nginx_request_body_json"
    tag_on_failure => "request_body_is_not_json"
}

Разобрать хедеры и перезаписать в то же поле 

kv {
   source => "nginx_request_headers"
   target => "nginx_request_headers"
}

Хедеры в логе выглядят так 

"nginx_request_headers": " host='elk.arturhaunt.ninja'  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 a2liYW5hYWRtaW46a2liYW5hYWRtaW4='  kbn-system-request='true' ...

После разбора: 

  "nginx_request_headers": {
    "accept": "*/*",
    "connection": "keep-alive",
    "accept-encoding": "gzip, deflate, br",
    "kbn-version": "7.14.0",
    "referer": "https://elk.domain.tld/app/discover",
    "accept-language": "en-gb",
    "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",
    "host": "elk.domain.tld",
    "content-type": "application/json",
    "kbn-system-request": "true",
    "authorization": "Basic a2liYW5hYWRtaW46a2liYW5hYWRtaW4="
  },
Источник — https://noname.com.ua/mediawiki/index.php?title=LogstashExample1&oldid=8924