ES Filebeat 使用 Pipeline 处理日志中的 @timestamp

推荐阅读

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"}

在 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                   # 如果遇到错误则忽略}}]
}

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

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

在 filebeat 中引用这个 pipeline：

filebeat.idle_timeout: 2s
filebeat.inputs:
- backoff: 1sbackoff_factor: 2close_inactive: 1henabled: trueencoding: plainharvester_buffer_size: 262144max_backoff: 10smax_bytes: 10485760paths:- /opt/trace.logscan_frequency: 10stail_lines: truetype: logfields:type: test-news-server
filebeat.name: filebeat-shiper
filebeat.spool_zie: 50000
output.elasticsearch:bulk_max_size: 8192hosts:- 10.11.16.211:30187- 10.11.16.212:30187- 10.11.16.213:30187- 10.11.16.214:30187- 10.11.16.215:30187index: test-news-timestampworkers: 4pipeline: "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:- messagemax_depth: 1overwrite_keys: true
setup.ilm.enabled: false
setup.template.name: test-news-timestamp-reverse
setup.template.pattern: test-news-timestamp-reverse-*

运行 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

如果不希望将 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"}}]
}

如果希望将 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"}}]
}

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

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

ES Filebeat 使用 Pipeline 处理日志中的 @timestamp相关推荐

FileBeat + Pipeline 解析日志保存至ElasticSearch（实战）
文章目录 FileBeat + Pipeline 解析日志保存至ElasticSearch(实战) 下载地址目的日志数据模拟Pipeline 创建pipeline 查看Pipeline是否创建 ...
canal 日志中 ERROR c.a.otter.canal.client.adapter.es.support.ESTemplate ...document missing 产生的可能的原因
canal 日志中经常会有日志信息 2019-08-14 15:02:24.843 [pool-2-thread-1] ERROR c.a.otter.canal.client.adapter.es. ...
es分页和mysql分页_用户logstash同步mysql数据到es中7.4.1版本以后输出的sql日志中没有分页信息...
问题描述: 在使用logstash同步mysql数据库中的数据,在使用logstash7.4.1以前的版本,输出的sql日志如下: [2019-12-09T10:50:01,585][INFO ][l ...
ELK 使用filebeat替代Logstash收集日志
使用beats采集日志之前也介绍过beats是ELK体系中新增的一个工具,它属于一个轻量的日志采集器,以上我们使用的日志采集工具是logstash,但是logstash占用的资源比较大,没有beat ...
filebeat+logstash收集错误日志发送邮件提醒
filebeat+logstash收集错误日志发送邮件提醒典型ELK应用架构因为只收集错误日志并且数据量并不是非常大所以简化流程使用filebeat+logstash发送异常日志软件版本说 ...
使用 Filebeat 对多行日志进行处理（multiline）
推荐阅读 Helm3(K8S 资源对象管理工具)视频教程:https://edu.csdn.net/course/detail/32506 Helm3(K8S 资源对象管理工具)博客专栏:https: ...
浅谈精准提取日志中的URL
背景在分析日志的时候发现有些日志中参数中包含其他的URL,例如: http://www.xxx.cn/r/common/register_tpl_shortcut.php?ico_url=http: ...
链路日志中追踪traceId
一,使用traceId概述平时出现问题查询日志是程序员的解决方式,日志一般会从服务器的日志文件,然后查找自己需要的日志,或者日志输出到es中,在es中进行搜索日志,可是对于目前流行的微服务或者单体服 ...
linux 统计日志数量总,shell统计日志中时间段内匹配的数量的方法
shell统计日志中时间段内匹配的数量的方法,有需要的朋友可以参考下. 假设日志文件mtasvr.log格式如下: T:24583088(04:02:06)[root:Info] 6KqowLDLAg ...

ES Filebeat 使用 Pipeline 处理日志中的 @timestamp

推荐阅读

Helm3（K8S 资源对象管理工具）视频教程：https://edu.csdn.net/course/detail/32506

Helm3（K8S 资源对象管理工具）博客专栏：https://blog.csdn.net/xzk9381/category_10895812.html

使用 Pipeline 处理日志中的 @timestamp

ES Filebeat 使用 Pipeline 处理日志中的 @timestamp相关推荐

最新文章

热门文章