当前位置:   article > 正文

ES & Filebeat 使用 Pipeline 处理日志中的 @timestamp_filebeat @ times

filebeat @ times

推荐阅读

Helm3(K8S 资源对象管理工具)视频教程:https://edu.csdn.net/course/detail/32506
Helm3(K8S 资源对象管理工具)博客专栏:https://blog.csdn.net/xzk9381/category_10895812.html

本文原文链接:https://blog.csdn.net/xzk9381/article/details/109535597,转载请注明出处。如有发现文章中的任何问题,欢迎评论区留言。

使用 Pipeline 处理日志中的 @timestamp

Filebeat 收集的日志发送到 ElasticSearch 后,会默认添加一个 @timestamp 字段作为时间戳用于检索,而日志中的信息会全部添加到 message 字段中,但是这个时间是 Filebeat 采集日志的时间,不是日志生成的实际时间,所以为了便于检索日志,需要将 @timestamp 替换为 message 字段中的时间。

这里使用的是 elasticseatch 提供的 pipeline 来进行替换。首先日志格式如下:

20-09-22 07:01:25.109 INFO - {"traceId":"65e97e88a61d7cd4558b8f3a203458fd"}
20-09-22 06:51:12.117 INFO - {"traceId":"4e0542c994919065f71536872ccb9677"}
  • 1
  • 2

在 Kibana 中的 Devtools 界面中编写如下 pipeline 并执行:

PUT _ingest/pipeline/test-news-server-online      # test-news-server-online 为流水线的名称
{
  "description": "test-news-server-online",       # 对 pipeline 进行描述
  "processors": [
    {
      "grok": {                                   # 使用 grok 对日志内容进行提取
        "field": "message",					      # 选择要提取信息的字段
        "patterns": [
          "%{TIMESTAMP_ISO8601:logatime}"         # 使用 TIMESTAMP_ISO8601 的标准匹配时间,将匹配的值赋值给新增的字段 logatime
        ],
        "ignore_failure": true		              # 如果日志中有不存在时间戳的行,可以添加这个配置来忽略匹配错误产生的 error 信息
      },
      "date": {					                  # 使用 data 时间戳插件来格式化时间输出,替代默认的 @timestamp
        "field": "logatime",				      # 指定使用新增的 logatime 字段
        "timezone": "Asia/Shanghai", 	          # 指定输出时间的时区,不指定的话可能会比正确的时间晚 8 个小时
        "formats": [
          "yy-MM-dd HH:mm:ss.SSS"	              # 指定时间输出的格式
        ],
        "ignore_failure": true		              # 如果遇到错误则忽略
      }
    }
  ]
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23

pipeline 编写完成后,在 Devtools 中可以使用如下命令进行查询:

GET _ingest/pipeline/test-news-server-online
  • 1

在 filebeat 中引用这个 pipeline:

filebeat.idle_timeout: 2s
filebeat.inputs:
- backoff: 1s
  backoff_factor: 2
  close_inactive: 1h
  enabled: true
  encoding: plain
  harvester_buffer_size: 262144
  max_backoff: 10s
  max_bytes: 10485760
  paths:
  - /opt/trace.log
  scan_frequency: 10s
  tail_lines: true
  type: log
  fields:
    type: test-news-server
filebeat.name: filebeat-shiper
filebeat.spool_zie: 50000
output.elasticsearch:
  bulk_max_size: 8192
  hosts:
  - 10.11.16.211:30187
  - 10.11.16.212:30187
  - 10.11.16.213:30187
  - 10.11.16.214:30187
  - 10.11.16.215:30187
  index: test-news-timestamp
  workers: 4
  pipeline: "test-news-server-online"				# 在此处指定 pipeline 的名称
processors:
- drop_fields:
    fields:
    - agent.ephemeral_id
    - agent.hostname
    - agent.id
    - agent.type
    - agent.version
    - ecs.version
    - input.type
    - log.offset
    - version
- decode_json_fields:
    fields:
    - message
    max_depth: 1
    overwrite_keys: true
setup.ilm.enabled: false
setup.template.name: test-news-timestamp-reverse
setup.template.pattern: test-news-timestamp-reverse-*
  • 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
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50

运行 filebeat,在 kibana 中查看日志信息,可以看到收集的日志信息中新增了 logatime 字段,@timestamp 字段的时间也与 logatime 字段保持了一致。

本文原文链接:https://blog.csdn.net/xzk9381/article/details/109535597

如果在 filebeat 运行的日志中发现了如下报错信息,有可能是日志中存在不含有时间戳的行(一般是由于日志被截断导致的,可以参考我的其他博客:使用 Filebeat 对多行日志进行处理(multiline)):

ERROR   pipeline/output.go:121  Failed to publish events: temporary bulk send failure
  • 1

如果不希望将 logatime 字段在日志中展示的话,可以将 pipeline 修改为如下内容:

PUT _ingest/pipeline/test-news-server-online
{
  "description": "test-news-server-online",
  "processors": [
    {
      "grok": {
        "field": "message",
        "patterns": [
          "%{TIMESTAMP_ISO8601:logatime}"
        ],
        "ignore_failure": true
      },
      "date": {
        "field": "logatime",
        "timezone": "Asia/Shanghai", 
        "formats": [
          "yy-MM-dd HH:mm:ss"
        ],
        "ignore_failure": true
      },
      "remove": {
        "field": "logatime"
      }
    }
  ]
}
  • 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

如果希望将 logatime 的值同时赋值给其他的新增字段,例如 realtime ,pipeline 修改如下:

PUT _ingest/pipeline/test-news-server-online
{
  "description": "test-news-server-online",
  "processors": [
    {
      "grok": {
        "field": "message",
        "patterns": [
          "%{TIMESTAMP_ISO8601:logatime}"
        ],
        "ignore_failure": true
      },
      "date": {
        "field": "logatime",
        "timezone": "Asia/Shanghai", 
        "target_field": "realtime"
        "formats": [
          "yy-MM-dd HH:mm:ss"
        ],
        "ignore_failure": true
      },
      "remove": {
        "field": "logatime"
      }
    }
  ]
}
  • 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

target_field 字段用于将一个值赋值给指定的字段,默认是给 @timestamp ,如果未提供该选项,则会默认更新 @timestamp 字段

本文原文链接:https://blog.csdn.net/xzk9381/article/details/109535597,转载请注明出处。如有发现文章中的任何问题,欢迎评论区留言。

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/小舞很执着/article/detail/812221
推荐阅读
相关标签
  

闽ICP备14008679号