filebeat采集日志到es并通过java读取日志

不同的日志格式需要不同的解析方式，这里的日志格式为时间开头的内容，我们就根据每个时间开头来分割日志，日志形式如下：

2021-09-30 14:21:31.962 [http-nio-50003-exec-9] ERROR com.elco.spring.advice.ExceptionAdvice - 业务异常
com.elco.core.exception.BusinessException: 该编码不存在
	at com.elco.eam.maindata.service.DictService.listDictByTypeCode(DictService.java:216)
2021-10-30 14:21:31.962 [http-nio-50003-exec-9] ERROR com.elco.spring.advice.ExceptionAdvice - 业务异常
com.elco.core.exception.BusinessException: 该字典不存在
	at com.elco.eam.maindata.service.DictService.listDictByTypeCode(DictS

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - C:UsersAdministratorDesktopfsdownload*.log
#  exclude_files: ['.gz$','INFO']
  multiline.type: pattern
  multiline.pattern: '^[0-9]{4}-[0-9]{2}-[0-9]{2}'
  #multiline.pattern: '^20'
  multiline.negate: true
  multiline.match: after
 
# 7.x的版本中需要禁用此索引生命周期，否则在指定es索引名字的时候会有问题
setup.ilm.enabled: false
# 添加模板配置，否则无法指定es的索引名
setup.template.name: "log"
setup.template.pattern: "log-*"
 
# 输出到es
output.elasticsearch:
  hosts: ["127.0.0.1:9200"]
  index: "my-log"
  #index: "log-%{+yyyy.MM.dd}"
processors:
  - script:
      lang: javascript
      id: my_filter
      tag: enable
      source: >
        function process(event){
          var str= event.Get("message");
          var time=str.split(" ").slice(0,2).join(" ");
          event.Put("start_time",time);
        }
  - timestamp:
      field: start_time 
    #  timezone: Asia/Shanghai
      layouts:
        - '2006-01-02 15:04:05'
        - '2006-01-02 15:04:05.999'
      test:
        - '2019-06-22 16:33:51'
        - '2019-11-18 04:59:51.123'
  - drop_fields:
      fields: [start_time]

参数说明：

1.multiline.pattern：

是按照年-月-日的形式进行分割的，也就是上一步骤中的日志形式，这里需要根据自身实际的日志来进行配置。

2.processor-script：

可选项。这里是为了获取日志内容中的时间作为es中@timestamp的值，也就是说如果没有这一步也可以，但是会导致es的@tempstamp的值为当前的时间而不是日志内容的时间，这样不利于进行日志时间范围的搜索。这里执行了一段js脚本，就是将日志内容的开头时间获取并赋值给start_time。

3.processor-timestamp：

可选项。这里将start_time赋值给@timestamp。

4.drop_fields：

可选项。删除start_time字段。

在kibana中执行查询索引的内容，结果如下：

 这里可以发现@timestamp和日志内容中开头的时间是一致的，这就是上一步骤中2，3，4所做的内容。如果没有2，3，4，这里的@timestamp的值就会是当前日志进入es的时间，不利于时间范围查询。

数据进入es之后就可以进行采集了，代码在github，欢迎star。