Анализ события JSON в Logstash - PullRequest
Новая
       13

Анализ события JSON в Logstash

0 голосов
/ 08 марта 2019

У меня логин в следующем формате, это обычный json с вложенными полями.

{
    "level": "info",
    "message": {
        "req": {
            "headers": {
                "host": "localhost:8080",
                "connection": "keep-alive",
                "x-forwarded-for": "192.168.1.1, 1.1.1.1",
                "x-forwarded-proto": "http"
            },
            "url": "/products?userId=493d0aec-a9a7-42a3",
            "method": "GET",
            "originalUrl": "/products?userId=493d0aec-a9a7-42a3",
            "params": {
                "0": "/products"
            },
            "query": {
                "userId": "493d0aec-a9a7-42a3"
            },
            "body": ""
        },
        "res": {
            "headers": {
                "traceid": "ac586e4e924048",
                "x-correlation-id": "57d7920d-b623-48f8",
                "content-type": "application/json;charset=UTF-8",
                "content-length": "2",
                "date": "Fri, 08 Mar 2019 09:55:45 GMT",
                "connection": "close"
            },
            "statusCode": 200,
            "body": "[]"
        },
        "gateway": "internal"
    },
    "correlationId": "57d7920d-b623-48f8",
    "timestamp": "2019-03-08T09:55:45.833Z"
}

Как правильно разобрать его, используя Filebeat и Logstash, чтобы увидеть все поля json в Kibana как отдельные (проанализированные) поля?У меня проблема с полем "message", в которое вложены поля json.У меня нет проблем, чтобы проанализировать событие, которое имеет строку в «сообщении», но не json.

Мои попытки:

1.Я пытался сказать Filebeat, что это json со следующей конфигурацией:
(и ничего не делать со стороны LS) 

filebeat.inputs:
- type: stdin
  json.keys_under_root: true
  json.add_error_key: true

Результат странный для меня, потому что я получил "message" в виде строкив Кибане, где все : заменены на =>

{
    "req" => {
        "originalUrl" => "/offers", "params" => {
            "0" => "/offers"
        }, "query" => {}, "body" => "", "headers" => {
            "accept-encoding" => "gzip", "user-agent" => "okhttp/3.8.1", "x-consumer-id" => "f2a6e4cd-2224-4535

Другие поля вне "сообщения" анализируются правильно

2.Я ничего не делал на стороне Filebeat и использую фильтр в LS: 

json {   
    source => "message"    
    target => "message_json"    
}

Журналы вообще не появляются в Kibana, я получил следующие ошибки в LS:

[2019-03-08T09: 55: 47,084] [WARN] [logstash.outputs.elasticsearch] Не удалось проиндексировать событие для Elasticsearch.{: status => 400,: action => ["index", {: _id => nil,: _index => "filebeat-6.5.0-2019.03.08-sdx",: _type => "doc",:routing => nil}, #],: response => {"index" => {"_ index" => "filebeat-6.5.0-2019.03.08-sdx", "_type" => "doc", " id "=>" ERS6XGkBgE -US7A6Mvt "," status "=> 400," error "=> {" type "=>" mapper_parsing_exception "," reason "=>" не удалось проанализировать поле [json.сообщение] типа [ключевое слово] "," reason_by "=> {" type "=>" invalid_state_exception "," reason "=>" Невозможно получить текст на START_OBJECT в 1: 461 "}}}}} [2019-03-08T09: 55: 47,485] [WARN] [logstash.outputs.elasticsearch] Не удалось проиндексировать событие для Elasticsearch.{: status => 400,: action => ["index", {: _id => nil,: _index => "filebeat-6.5.0-2019.03.08-sdx",: _type => "doc",:routing => nil}, #],: response => {"index" => {"_ index" => "filebeat-6.5.0-2019.03.08-sdx", "_type" => "doc", " id "=>" EhS6XGkBgE -US7A6Mvt "," status "=> 400," error "=> {" type "=>" mapper_parsing_exception "," reason "=>" не удалось проанализировать поле [json.сообщение] типа [ключевое слово] "," reason_by "=> {" type "=>" invalid_state_exception "," reason "=>" Невозможно получить текст на START_OBJECT в 1: 461 "}}}}}

Этот фильтр отлично работает для меня, если поле "message" является строкой (не json).

Есть идеи, как проанализировать вложенный json в "поле сообщения"?

1 Ответ

0 голосов
/ 14 марта 2019

У меня проблемы с анализом json с помощью logstash.

Я боролся с этой проблемой некоторое время.И не удалось решить в logstash.

Но, к счастью, у нас есть загруженный узел в самом эластичном поиске.

Я хочу предложить свое решение вашей проблемы:

Вы создаете конвейер (очень простой конвейер): 

{
    "description": "Parse JSON log",
    "processors": [
      {
        "json": {
          "field": "message",
          "target_field": "message-json"
        }
      },
      {
        "remove": {
          "field": "message"
        }
      },
      {
        "append": {
          "field": "tags",
          "value": [
            "isjsonlog"
          ]
        }
      }
    ],
    "on_failure": [
      {
        "append": {
          "field": "tags",
          "value": [
            "nonjsonlog"
          ]
        }
      }
    ]
  }

В своем выходном плагине вы настраиваете: 

elasticsearch {
    hosts => [ localhost ]
    index => "filebeat-%{+YYYY.MM.dd}"
    manage_template => false
    pipeline => "your_pipeline_name"
  }

И вы забываете проблемы с синтаксическим анализом json.

Если вы используете filebeat , вы можете отправлять журналы json напрямую в конвейер , настроив filebeat: 

 output.elasticsearch:
    ....
    pipelines:
       - pipeline: "your pipeline name"
          when:
            contains:
               message: "{"
...