如何使用 FileBeat 将管道分隔格式的日志数据以 JSON 格式发送到 Elasticsearch？答案

【问题标题】：How do I use FileBeat to send log data in pipe separated format to Elasticsearch in JSON format?如何使用 FileBeat 将管道分隔格式的日志数据以 JSON 格式发送到 Elasticsearch？
【发布时间】：2017-08-07 10:01:43
【问题描述】：

我正在监控的日志文件具有以下格式的日志：

Oct 23 16:06:44 server smbd_audit: user01|192.168.0.23|project|opendir|ok|.
Oct 23 16:06:44 server smbd_audit: user01|192.168.0.23|project|closedir|ok|
Oct 23 16:06:44 server smbd_audit: user01|192.168.0.23|project|open|ok|r|file.txt
Oct 23 16:06:44 server smbd_audit: user01|192.168.0.23|project|pread|ok|file.txt
Oct 23 16:06:44 server smbd_audit: user01|192.168.0.23|project|close|ok|file.txt

如何在使用 FileBeat 将数据发送到 Elasticsearch 之前对其进行格式化？

我希望我的文档如下所示（不包括 elasticsearch 元数据字段）：

{
  "timestamp": "Oct 23 16:06:44",
  "machine-name": "server",
  "type": "smbd_audit",
  "username": "user01",
  "machine-ip": "192.168.0.23",
  "directory": "project",
  "operation": "opendir",
  "success": "ok",
  "file": "file.txt"
}

【问题讨论】：

标签： elasticsearch logstash kibana elastic-stack filebeat

【解决方案1】：

我假设你不想使用 Logstash，所以你可以使用 ingest pipeline 和 Grok。

put _ingest/my-pipeline    
{
  "description": "My Ingest Pipeline",
  "processors": [
    {
      "grok": {
        "field": "message",
        "patterns": [
          "%{SYSLOGTIMESTAMP:log_date} %{WORD:machine-name} %{WORD:type}: %{WORD:username}|{IP:machine-ip}|{WORD:directory}|{WORD:operation}|{WORD:success}|{WORD:file}"
        ]
      }
    },
    {
      "date": {
        "field": "log_date"
      }
    }
  ]
}

完全未经测试，但至少应该给你一些东西。

【讨论】：

感谢 @dwjv 向我介绍摄取管道。
没问题，仅供参考，底部的“日期”部分是为了让您可以使用日志中的日期/时间进行搜索（在 Kibana 等中），而不是默认的摄取日期.未经测试，您可能需要添加格式以便 ES 可以正确翻译它。例如"date":{"field":"log_date","formats": ["MMM dd HH:mm:ss"]}