需求描述

使用 Filebeat 从 log 文件中采集 JSON 格式的日志，发送到 ES 中，并在 ES 中显示 JSON 日志的各字段和数据。

问题一：如何采集 JSON 格式的日志

在 filebeat.yml 文件中进行相应的配置：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
- type: log
  enabled: true

  paths:
    - E:\testjson.log
  processors:
  - script:
      lang: javascript
      source: >
        function process(event) {
            var message = event.Get("message");
            message = message.replace(/\\x22/g,'"');
            message = message.replace(/\,-/g,'');
            event.Put("message", message.trim());
        }
  - decode_json_fields:
      fields: ["message"]
      # 是否解析数组
      process_array: false
      max_depth: 1
      # 配置所有解析出来的字段放在哪个 key 上（空字符串则是放根节点）
      target: ""
      overwrite_keys: false
      add_error_key: true
  # `decode_json_fields` 和 `json` 二选一
  #json:
  #  # 默认这个值是 FALSE 的，也就是我们的 json 日志解析后会被放在 json 键上。设为 TRUE，所有的 keys 就会被放到根节点
  #  keys_under_root: true
  #  # 是否要覆盖原有的 key，这是关键配置，将 keys_under_root 设为 TRUE 后，再将 overwrite_keys 也设为 TRUE，就能把 filebeat 默认的 key 值给覆盖了
  #  overwrite_keys: true
  #  # 添加 json_error key 键记录：json 解析失败错误
  #  add_error_key: true
  #  # 指定 json 日志解析后放到哪个 key 上，默认是 json，你也可以指定为 log 等。该字段必须处于 json 的 root
  #  message_key: message

具体配置可参考相应 文档 json 部分。

注意：每条 JSON 日志只能占一行，如果换行的话会解析出错！

问题：JSON 格式解析报错 "Error decoding JSON: EOF" 以及 "key not found"

具体错误如下：

1
2
2020-12-14T16:01:50.789+0800    ERROR   [reader_json]   readjson/json.go:57     Error decoding JSON: EOF
2020-12-14T16:01:50.789+0800    DEBUG   [processors]    processing/processors.go:128    Fail to apply processor global{timestamp=[field=start_time, target_field=@timestamp, timezone=Asia/Shanghai, layouts=[2006-01-02 15:04:05.999]], drop_fields={"Fields":["log","host","input","agent","ecs","start_time"],"IgnoreMissing":false}}: failed to get time field start_time: key not found

错误原因：刚开始报了上述错误，只有第一条日志发出去了，后边的日志都会报错 Error decoding JSON: EOF，打开 debug 日志还会发现 JSON 日志的第一个字段会 "key not found"。原因是 JSON 格式的日志每条最后必须加逗号 ,，而我的日志中仅将其换行。

解决方法：JSON 日志的大括号后加逗号

1
2
{json日志},
{json日志},

问题二：如何发送到 ElasticSearch

在 filebeat.yml 中设置 output.elasticsearch：

1
2
3
output.elasticsearch:
  # Array of hosts to connect to.
  hosts: ["ip:9200"]

这种设置之后，日志存入 ES 的 index 默认是 "filebeat-%{[agent.version]}-%{+yyyy.MM.dd}"。

可以通过 index 参数设置插入索引：

1
2
3
4
5
6
7
8
9
10
output.elasticsearch:
  # Array of hosts to connect to.
  hosts: ["IP:9200"]
  # 修改的配置如下，这里的 %{+yyyy.MM.dd} 取得是 @timestamp 字段
  index: "test-%{[agent.version]}-normal-%{+yyyy.MM.dd}"
setup.template.overwrite: true
# 如果需要修改 index，以下两个属性必须配置
setup.template.name: "test"
setup.template.pattern: "test-*"
setup.ilm.enabled: false

也可以按照 官方文档 动态配置索引。

可能出现如下错误：

1
failed to publish events: temporary bulk send failure

问题三：如何用自己的时间戳替换 @timestamp

配合使用 timestamp 处理器 和 drop_fields 处理器。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
processors: 
  - timestamp:
      # 格式化时间值给时间戳
      field: start_time
      # 使用我国东八区时间格式化 log 时间，filebeat 默认是 UTC+0，如果没改 filebeat 时区，那么这里设置东 8 的话会让你的 timestamp 减去 8 个小时
      timezone: Asia/Shanghai
      layouts:
        # 具体日期格式可参考文档中的说明
        - '2006-01-02 15:04:05.999'
      test: 
        # test 字段：必须能够成功解析的日期内容
        - '2019-06-22 16:33:51.111'
  - drop_fields:
      fields: ["log","host","input","agent","ecs","start_time"]
      ignore_missing: false

从 start_time 字段中提取日志时间，并放入时间戳中，并在 drop_fields 中将该字段删掉即可。

整体配置文件如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - E:\testjson.log
  json:
    keys_under_root: true
    overwrite_keys: true
    add_error_key: true
    message_key: message
output.elasticsearch:
  # Array of hosts to connect to.
  hosts: ["192.168.73.101:9200"]
processors:
  - timestamp:
      # 格式化时间值给时间戳
      field: start_time
      # 使用我国东八区时间格式化 log 时间
      timezone: Asia/Shanghai
      layouts:
        - '2006-01-02 15:04:05.999'
      test:
        - '2019-06-22 16:33:51.111'

  - drop_fields:
      fields: ["log","host","input","agent","ecs","start_time"]
      ignore_missing: false

最终效果

原始日志：

1
2
{"start_time": "2020-12-13 10:37:01.072","type": "CsbTest","level": "INFO","message":"数据1","parameter":["6677"]},
{"start_time": "2020-12-13 10:37:01.072","type": "CsbTest","level": "INFO","message":"数据2","parameter":["12121"]},

ES 存储日志：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
{
  "_index" : "filebeat-7.10.0-2020.12.14-000001",
  "_type" : "_doc",
  "_id" : "8d1FYHYBjbaP5OPrGcZL",
  "_score" : 1.0,
  "_source" : {
    "@timestamp" : "2020-12-14T02:37:01.072Z",
    "message" : "数据2",
    "parameter" : [
      "6677"
    ],
    "type" : "CsbTest",
    "level" : "INFO",
    "fields" : {
      "log_type" : "normal"
    }
  }
}

参考