在 Lostash 的 beats.config 中自定义来自 filebeat 的日志答案

【问题标题】：Customize logs from filebeat in the Lostash's beats.config在 Lostash 的 beats.config 中自定义来自 filebeat 的日志
【发布时间】：2018-10-29 12:11:54
【问题描述】：

我将 ELK 与 filebeat 一起使用。我正在将日志从 filebeat 发送到 Logstash，然后从那里发送到 Elastic 并在 Kibana 中进行可视化。我正在粘贴显示在 kibana 日志结果中的 json 结果，如下所示：

    {
  "_index": "filebeat-6.4.2-2018.10.30",
  "_type": "doc",
  "_source": {
    "@timestamp": "2018-10-30T09:15:31.697Z",
    "fields": {
      "server": "server1"
    },
    "prospector": {
      "type": "log"
    },
    "host": {
      "name": "kushmathapa"
    },
    "message": "{ \"datetime\": \"2018-10-23T18:04:00.811660Z\", \"level\": \"ERROR\", \"message\": \"No response from remote. Handshake timed out or transport failure detector triggered.\" }",
    "source": "C:\\logs\\batch-portal\\error.json",
    "input": {
      "type": "log"
    },
    "beat": {
      "name": "kushmathapa",
      "hostname": "kushmathapa",
      "version": "6.4.2"
    },
    "offset": 0,
    "tags": [
      "lighthouse1",
      "controller",
      "trt"
    ]
  },
  "fields": {
    "@timestamp": [
      "2018-10-30T09:15:31.697Z"
    ]
  }
}

我希望它显示为

    {
  "_index": "filebeat-6.4.2-2018.10.30",
  "_type": "doc",
  "_source": {
    "@timestamp": "2018-10-30T09:15:31.697Z",
    "fields": {
      "server": "server1"
    },
    "prospector": {
      "type": "log"
    },
    "host": {
      "name": "kushmathapa"
    },
    "datetime": 2018-10-23T18:04:00.811660Z,
    "log_level": ERROR,
    "message": "{ \"No response from remote. Handshake timed out or transport failure detector triggered.\" }",
    "source": "C:\\logs\\batch-portal\\error.json",
    "input": {
      "type": "log"
    },
    "beat": {
      "name": "kushmathapa",
      "hostname": "kushmathapa",
      "version": "6.4.2"
    },
    "offset": 0,
    "tags": [
      "lighthouse1",
      "controller",
      "trt"
    ]
  },
  "fields": {
    "@timestamp": [
      "2018-10-30T09:15:31.697Z"
    ]
  }
}

我的 beats.config 现在看起来像这样

  input {
  beats {
    port => 5044
  }
}

output {
  elasticsearch {
    hosts => "localhost:9200"
    manage_template => false
    index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}" 
  } stdout {
    codec => rubydebug { metadata => true }
  }
}

我已经应用了过滤器，但我似乎遗漏了一些东西。

【问题讨论】：

标签： logstash elastic-stack logstash-grok filebeat

【解决方案1】：

您可以使用看起来像这样的配置文件。在 grok 过滤器中，将要摄取的日志格式添加到 elasticsearch（例如参考上述配置）。

input {
beats {
port => 5044
id => "my_plugin_id"
tags => ["logs"]
type => "abc"
}
}
filter {
if [type] == "abc" {
 mutate {
    gsub => [ "message", "\r", "" ]
}

    grok {
        break_on_match => true
                match => {
                         "message" => [
                         "%{TIMESTAMP_ISO8601:timestamp}%{SPACE}%{LOGLEVEL:log_level}%{SPACE}%{GREEDYDATA:message}"
                         ]
                  }
                  overwrite => [ "message" ]
    }

    grok {
        break_on_match => true
                match => {
                         "message" => [
                          "%{TIMESTAMP_ISO8601:timestamp}%{SPACE}%{LOGLEVEL:log_level}%{SPACE}%{GREEDYDATA:message}"
                         ]
                  }
                  overwrite => [ "message" ]
    }

date {
   match => [ "timestamp" , "yyyy-MM-dd HH:mm:ss,SSS" ]
} 
}
}
output {
if [type] == "abc" {
elasticsearch { 
hosts => ["ip of elasticsearch:port_number of elasticsearch"]
index => "logfiles"
} 
}
else {
elasticsearch { 
hosts => ["ip of elasticsearch:port_number of elasticsearch"]
index => "task_log"
} 
}
stdout {
codec => rubydebug { metadata => true }
}
}

【讨论】：

由于消息是 json 格式（问题中的更新），grok 过滤器不起作用，因为消息是 json 格式
参考这个stackoverflow.com/questions/38869886/… 我想你会对你更新的问题有所了解。

【解决方案2】：

Logstash 需要知道您收到的 message 字段是 JSON 格式。您可以在此处使用json 过滤器并立即获得您要查找的几乎所有内容：

filter {
    json {
        target => "message"
    }
}

如果需要，您可以使用突变或添加/删除字段将 level 重命名为 log.level 和 datetime 重命名为 @datetime。

【讨论】：