ELKstack

Logstash用日志时间替换@timestamp, 日期修改

缘由

因为最近安全要求越来越高,服务器日常的操作权限很难申请,所以把尘封一年之久的Elastic stack日志收集分析系统重新利用起来。

目前我导入了java日志、apache日志、nginx日志以及mysql的相关日志。

技术方案


我选择的技术方案是filebeat做客户端推送日志到redis的list,logstash从redis定时拉取日志再根据配置文件进行处理,web端kibana配合elasticsearch分析和展示日志。
这里很多用户都会遇到一个问题,因为filebeat和logstash都是分批次每隔几秒去推送和拉取日志,logstash给日志打的时间戳是从redis里面拉取的时刻,和日志本身的时间戳不一样,导致网页默认排序会造成日志顺序错乱,虽然可以通过对offset排序得到正常的日志顺序,但是多少和用时间戳排序感觉不一样,会让强迫症患者很不爽。这里我的解决办法是通过logstash的date filter插件实现用日志时间戳替换logstash的@timestamp字段,这样默认的排序就是按时间先后的。

date filter插件是自带的,基本原理很简单:用logstash的grok正则匹配出日志中的时间字段,然后用时间字段替换@timestamp.

具体配置示例

  • filebeat.yml:


这里提一下吧,yml配置文件一定要注意缩进,和python一样,不然会有问题

  1. filebeat.inputs:
  2. - input_type: log
  3.   enabled: true
  4.   encoding: gbk
  5.   scan_frequency: 10s
  6.   paths:
  7.     - D:\logs\task_dev\system.log
  8.   fields:
  9.     list: test-0138-task-system
  10.   multiline.pattern: '^[\d][\d]:[\d][\d]:[\d][\d]\.[\d]{3}'
  11.   multiline.negate: true
  12.   multiline.match: after
  13. output.redis:
  14.   hosts: ["your_redis_server:your_redis_host"]
  15.   db: 0
  16.   timeout: 5
  17.   key: "default_list"
  18.   keys:
  19.     - key: "%{[fields.list]}"
  20.       mapping:
  21.       "test-0138-task-system""test-0138-task-system"
  • logstash配置 elk.conf:


这里注意datetime的format里毫秒不是fff而是大写的SSS,写错会导致logstash假死

  1. # 从redis拉取
  2. input {
  3.   redis {
  4.     type => "test-0138-task-system"
  5.     host => "your_redis_server"
  6.     port => "your_redis_port"
  7.     db => "0"
  8.     data_type => "list"
  9.     key => "test-0138-task-system"
  10.   }
  11. }
  12. filter {
  13.         if [type] =~ "test-98" { # =~运算符是正则匹配,我这里的意思是type里面包含"test-98"
  14.                 grok {  # grok是匹配正则插件,grok自带很多内置的pattern,也可以自己写正则匹配
  15.                         match => ["message", "%{TIMESTAMP_ISO8601:logdate}"]  # message是日志的内容,TIMESTAMP_ISO8601是grok内置的日志匹配pattern之一,主要匹配形如"2019-08-21 11:26:09.717"这样的日期格式,匹配之后把日期赋值给自定义的logdate变量
  16.                 }
  17.                 date {
  18.                         match => ["logdate", "yyyy-MM-dd HH:mm:ss.SSS"]  # 这里是大坑,一般毫秒是fff,但是这里是SSS,写fff的话会报错导致logstash假死
  19.                         target => "@timestamp"  # 把匹配出的时间戳赋值给logstash的timestamp
  20.         }
  21.         } else if [type] =~ "apache" {  # 这里是匹配我的apache日志
  22.                 grok {
  23.                         match => ["message", "%{HTTPDATE:logdate}"]  # HTTPDATE也是内置pattern,匹配形如"21/Aug/2019:15:56:09 +0800"的日志
  24.                 }
  25.                 date {
  26.                         match => ["logdate", "dd/MMM/yyyy:HH:mm:ss Z"]  # Z是时区,匹配这里的+0800,中国的东八区
  27.                         target => "@timestamp"
  28.         }
  29.         }
  30. }
  31. output {
  32.     elasticsearch {
  33.       hosts => ["127.0.0.1:9200"]
  34.       index => "%{[fields][list]}-%{+YYYY.MM.dd}"
  35.     }
  36. }

不同的应用日志时间格式可能不一样,具体自己的格式是否有内置pattern可以参考:
grok在线调试及内置pattern查询
logstash支持时间戳格式配置可以参考:
Date filter plugins 指南

用ruby日期字段修改

我们有些很老的java应用时间戳只有时分秒,没有日期,我这里使用ruby脚本把日期加上:

  1. if [type] =~ "prod-228" {  
  2.                 grok {  
  3.                         match => ["message""%{TIME:logdate}"]  
  4.                 }  
  5.                 ruby {  
  6.                         code => " 
  7.                         event.set('stock_date', Time.now.strftime('%Y-%m-%d') + ' ' + event.get('logdate')) 
  8.                         "  
  9.                 }  
  10.                 date {  
  11.                         match => ["stock_date""yyyy-MM-dd HH:mm:ss.SSS"]  
  12.                         target => "@timestamp"  
  13.                 }  
  14.                   
  15. }  

其它参考链接:
Logstash手册
Logstash官方配置example

有任何问题欢迎留言,如果帮助到你请移步我的about关注我的youtube channel,谢谢。

(1)

本文由 永烁星光 作者:linus 发表,转载请注明来源!

关键词:, ,

热评文章

发表评论