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

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

使用 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: 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-*

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

如果在 filebeat 运行的日志中发现了如下报错信息，有可能是日志中存在不含有时间戳的行（一般是由于日志被截断导致的，可以参考处理多行日志的文档）：

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 字段

ES & Filebeat 使用 Pipeline 处理日志中的 @timestamp的更多相关文章

Filebeat和logstash 使用过程中遇到的一些小问题记录
一.filebeat 收集软链文件日志 1.1.场景由于我们新部署的Nginx 日志都是采用的软链的形式. lrwxrwxrwx 1 root root 72 Apr 6 00:00 jy.baid ...
ELK之filebeat替代logstash收集日志
filebeat->redis->logstash->elasticsearch 官网下载地址:https://www.elastic.co/downloads/beats/file ...
【ElasticSearch】使用AWS云ES服务来分析程序日志
最近公司系统升级,有些API的调用接口达到了每天10几万的请求量.目前公司里的日志,都是写文本文件中的.为了能够更好的分析这些日志数据,公司采用了AWS 的 ElasticSearch服务来分析日志. ...
ELK 使用filebeat替代Logstash收集日志
使用beats采集日志之前也介绍过beats是ELK体系中新增的一个工具,它属于一个轻量的日志采集器,以上我们使用的日志采集工具是logstash,但是logstash占用的资源比较大,没有beat ...
使用kibana可视化报表实时监控你的应用程序，从日志中找出问题，解决问题
先结果导向,来看我在kibana dashborad中制作的几张监控图. 一:先睹为快 dashboard1:监控几个维度的日志,这么点日志量是因为把无用的清理掉了,而且只接入了部分应用. <1 ...
转 -Filebeat + Redis 管理 LOG日志实践
Filebeat + Redis 管理 LOG日志实践小赵营关注 2019.01.06 17:52* 字数 1648 阅读 24评论 0喜欢 2 引用转载请注明出处某早上,领导怒吼声远远传来 ...
ElasticSearch+Logstash+Filebeat+Kibana集群日志管理分析平台搭建
一.ELK搜索引擎原理介绍在使用搜索引擎是你可能会觉得很简单方便,只需要在搜索栏输入想要的关键字就能显示出想要的结果.但在这简单的操作背后是搜索引擎复杂的逻辑和许多组件协同工作的结果. 搜索引擎的组 ...
ELK快速入门（四）filebeat替代logstash收集日志
ELK快速入门四-filebeat替代logstash收集日志 filebeat简介 Filebeat是轻量级单用途的日志收集工具,用于在没有安装java的服务器上专门收集日志,可以将日志转发到log ...
ELK之elasticsearch插件导致filebeat没有上传日志至elasticsearch解决办法
使用filebeat收集nginx发现日志为上传,elasticsearch没有日志,kibana没有展示查看filebeat日志日志目录为/var/log/filebeat 下面有多个日志文件 ...

随机推荐

[vijos P1512] SuperBrother打鼹鼠
这周好好码树状数组和线段树!!之前没写过二维树状数组,凭借一维的思路居然写了个比较像模像样的东西出来,原来我没那么脑残.唯一要注意的就是getsum四个矩形加减的边界条件,这里看了别人标程才意识到错误 ...
log file sync
Recently, our application system has updated one app. I receive a email of complain the db server ch ...
MySQL之最基本命令
前言:以下是数据库最基础最常用的命令,特别适用初学者练习,希望通过不断练习这些命令来熟练操作.巩固基础,因为只有不断地练习才能将知识真正变成自己的东西. 快速查看以下内容: 操作命令创建数据库 C ...
Java经典编程题50道之二十五
一个5位数,判断它是不是回文数.即12321是回文数,个位与万位相同,十位与千位相同. public class Example25 { public static void main(Stri ...
APIPA
自动专用IP地址(Automatic Private IP Address,APIPA)是当客户端无法从DHCP服务器中获得IP地址时自动配置的地址.IPv4地址前缀169.254/16已经被IANA ...
使用SQL SERVER 来自动发送邮件
可以使用SQL SERVER 来发送自动邮件,主要是使用SQL SERVER 的dbo.sp_send_dbmail 存储过程(在msdb数据库中). 具体步骤如下: Step1: 编写要发送的邮件内 ...
SQL查询临时表空间的数据
史上最简单OpenGL+VS2017环境配置
这里采用最简单的方法: (1)添加Nuget包管理器(方便我们为项目添加OpenGL的库) (2)新建一个项目右键我们的项目名 ——> 管理NuGet程序包——>搜索nupengl,然 ...
css 文本溢出
多行文本溢出处理: display: -webkit-box; -webkit-box-orient: vertical; -webkit-line-clamp: 3; // 3 行 overflow ...
python多线程与GIL（转）
作者:卢钧轶(cenalulu) 本文原文地址:http://cenalulu.github.io/python/gil-in-python/ GIL是什么 GIL(Global Interprete ...