logstash/filebeat只接收最近一段时间的数据

项目整体架构为filebeat+logstash+elasticsearch，logstash版本为：7.17.3，filebeat版本为：7.16.3

需求

因为项目是通过logstash按天创建es索引，但服务器内部日志过多，如果启动的话，会将所有的日志建立索引，导致索引过多，从而数据过多，目前我想把7天内的日志推到es中，通过logstash，我该怎么做？

当然可以通过linux定时任务删除日志可以实现，但不想这么做，仅仅想通过logstash做筛选。

方法

采用logstash的filter，并使用ruby语句（ruby语句必须在filter范围内）

过滤事件：

filter{
	ruby {
		code => "event.cancel if event['message'] =~ /^info/ " #正则匹配message字段以info开头，那么去除这个事件，不继续往下走
	}
}
1
2
3
4
5

过滤7天内的数据：

filter{
    ruby {
      code => "event.cancel if (Time.now.to_f - event.get('@timestamp').to_f) > (60 * 60 * 24 * 5)"
    }	
}
# 其中，event['@timestamp']不行，会报错，必须写为event.get('@timestamp')
1
2
3
4
5
6

filebeat方式

filebeat的配置文件中包含这么一个参数：ignore_older

如果启用，那么Filebeat会忽略在指定的时间跨度之前被修改的文件。如果你想要保留日志文件一个较长的时间，那么配置ignore_older是很有用的。例如，如果你想要开始Filebeat，但是你只想发送最近一周最新的文件，这个情况下你可以配置这个选项。

你可以用时间字符串，比如2h（2小时），5m（5分钟）。默认是0，意思是禁用这个设置。

你必须设置ignore_older比close_inactive更大。

close_inactive

当启用此选项时，如果文件在指定的持续时间内未被获取，则Filebeat将关闭文件句柄。当harvester读取最后一行日志时，指定周期的计数器就开始工作了。它不基于文件的修改时间。如果关闭的文件再次更改，则会启动一个新的harvester，并且在scan_frequency结束后，将获得最新的更改。

推荐给close_inactive设置一个比你的日志文件更新的频率更大一点儿的值。例如，如果你的日志文件每隔几秒就会更新，你可以设置close_inactive为1m。如果日志文件的更新速率不固定，那么可以用多个配置。

将close_inactive设置为更低的值意味着文件句柄可以更早关闭。然而，这样做的副作用是，如果harvester关闭了，新的日志行不会实时发送。

关闭文件的时间戳不依赖于文件的修改时间。代替的，Filebeat用一个内部时间戳来反映最后一次读取文件的时间。例如，如果close_inactive被设置为5分钟，那么在harvester读取文件的最后一行以后，这个5分钟的倒计时就开始了。

你可以用时间字符串，比如2h（2小时），5m（5分钟）。默认是5m。

scan_frequency

Filebeat多久检查一次指定路径下的新文件（PS：检查的频率）。例如，如果你指定的路径是 /var/log/* ，那么会以指定的scan_frequency频率去扫描目录下的文件（PS：周期性扫描）。指定1秒钟扫描一次目录，这还不是很频繁。不建议设置为小于1秒。

如果你需要近实时的发送日志行的话，不要设置scan_frequency为一个很低的值，而应该调整close_inactive以至于文件处理器保持打开状态，并不断地轮询你的文件。

默认是10秒。

画外音：

这里再重点说一下 ignore_older , close_inactive , scan_frequency 这三个配置项

• ignore_older： 它是设置一个时间范围（跨度），不在这个跨度范围之内的文件更新都不管
• scan_frequency： 它设置的是扫描文件的频率，看看文件是否更新
• close_inactive：它设置的是文件如果多久没更新的话就关闭文件句柄，它是有一个倒计时，如果在倒计时期间，文件没有任何变化，则当倒计时结束的时候关闭文件句柄。不建议设置为小于1秒。

如果文件句柄关了以后，文件又被更新，那么在下一个扫描周期结束的时候变化发现这个改变，于是会再次打开这个文件读取日志行，前面我们也提到过，每个文件上一次读到什么位置（偏移量）都记录在registry文件中

结果配置为：

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /home/logs/monitor/*.log
  ignore_older: 168h
  close_inactive: 24h
  fields:
    type: beats-monitor-store
- type: log
  enabled: true
  paths:
    - /home/logs/gateway/*.log
  ignore_older: 168h
  close_inactive: 24h
  fields:
    type:  beats-gateway-store
output.logstash:
  hosts: ["xxxx:5044"]

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

参考文献：

Filebeat 模块与配置