Filebeat将mysql-slow.log的几行合并为一行

Question

我正在尝试使用Filebeat Logstash和Elasticsearch分析mysql-slow.log。我在mysql-slow.log文件中有如下消息：

# Time: 2019-11-08T20:02:05.474508Z
# User@Host: user[user] @ localhost []  Id:     2
# Query_time: 0.000716  Lock_time: 0.000223 Rows_sent: 2  Rows_examined: 2
SET timestamp=1573243325;
select * from Persons;

首先，我尝试使Filebeat将这行日志消息以5行的形式发送给elasticsearch，但所有这些都以一行的形式一起发送。

我在filebeat.yml中设置了多行输入

multiline.pattern = `^\#`
multiline.negate = true
multiline.match = after

很遗憾，它不起作用，elasticsearch分别接收行

1. 消息->＃时间：2019-11-08T20：02：05.474508Z
2. 消息->＃用户@主机：用户[用户] @本地主机[] ID：2等等...

我想通过以下格式在一封邮件中收到它：

# Time: 2019-11-08T20:02:05.474508Z # User@Host: user[user] @ localhost []  Id:     2 # Query_time: 0.000716  Lock_time: 0.000223 Rows_sent: 2  Rows_examined: 2 SET timestamp=1573243325; select * from Persons;

Answer 1

您的多行模式是错误的，它将匹配以#开头的任何行，因此示例中的前三行将是filebeat / logstash的事件。

您需要更改多行模式以仅匹配事件的第一行，即以# Time开头的行。

以下文件拍子配置可用于我的测试。

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /opt/data/stackoverflow/*.log
  multiline.pattern: '^\#[[:space:]]Time'
  multiline.negate: true
  multiline.match: after

output.logstash:
  hosts: ["elk:5044"]

logstash管道简单监听5044并输出到elasticsearch，然后结果如下。

正如您所看到的，所有文件行都被索引为elasticsearch上的单个事件。