Flume中的TaildirSource

本是古典何须时尚 2022-05-12 18:18 177阅读 0赞

在通过Flume收集日志的业务场景中，一般都会遇到下面的情况，在日志收集服务器的某个目录下，会按照一段时间生成一个日志文件，并且日志会不断的追加到这个文件中，比如，每小时一个命名规则为log\_20151015\_10.log的日志文件，所有10点产生的日志都会追加到这个文件中，到了11点，就会生成另一个log\_20151015\_11.log的文件。

这种场景如果通过flume（1.6）收集，当前提供的Spooling Directory Source和Exec Source均不能满足动态实时收集的需求，在当前正在开发的flume1.7版本中，提供了一个非常好用的TaildirSource，使用这个source，可以监控一个目录，并且使用正则表达式匹配该目录中的文件名进行实时收集。

我将TaildirSource的相关源码下载下来（需要做简单修改），然后集成到Flume1.6中，满足了上面提到的需求，获得了良好的效果。

源码下载地址： [点击下载][Link 1]

将源码单独编译，打成jar包，上传到$FLUME\_HOME/lib/目录下。

下面的例子中，通过flume监控/tmp/lxw1234-flume/目录下，命名规则为log\_.\*.log的文件，并将文件内容实时的写入/tmp/flumefiles/目录下，即：source为TaildirSource，sink为file\_roll；

## Agent配置 ##

agent\_lxw1234的配置文件如下（$FLUME\_HOME/conf/agent\_lxw1234\_conf.properties）：

1.  \#-->设置sources名称
2.  agent\_lxw1234.sources = sources1
3.  \#--> 设置channel名称
4.  agent\_lxw1234.channels = fileChannel
5.  \#--> 设置sink 名称
6.  agent\_lxw1234.sinks = sink1
7.   
8.  \# source 配置
9.  agent\_lxw1234.sources.sources1.type = com.lxw1234.flume17.TaildirSource
10. agent\_lxw1234.sources.sources1.positionFile = /tmp/flume/taildir\_position.json
11. agent\_lxw1234.sources.sources1.filegroups = f1
12. agent\_lxw1234.sources.sources1.filegroups.f1 = /tmp/lxw1234-flume/log\_.\*.log
13. agent\_lxw1234.sources.sources1.batchSize = 100
14. agent\_lxw1234.sources.sources1.backoffSleepIncrement = 1000
15. agent\_lxw1234.sources.sources1.maxBackoffSleep = 5000
16. agent\_lxw1234.sources.sources1.channels = fileChannel
17.  
18. \# sink1 配置
19. agent\_lxw1234.sinks.sink1.type = file\_roll
20. agent\_lxw1234.sinks.sink1.sink.directory = /tmp/flumefiles
21. agent\_lxw1234.sinks.sink1.sink.rollInterval = 0
22. agent\_lxw1234.sinks.sink1.channel = fileChannel
23.  
24. \# fileChannel 配置
25. agent\_lxw1234.channels.fileChannel.type = file
26. \#-->检测点文件所存储的目录
27. agent\_lxw1234.channels.fileChannel.checkpointDir = /opt/flume/checkpoint/lxw1234/
28. \#-->数据存储所在的目录设置
29. agent\_lxw1234.channels.fileChannel.dataDirs = /opt/flume/data/lxw1234/
30. \#-->隧道的最大容量
31. agent\_lxw1234.channels.fileChannel.capacity = 10000
32. \#-->事务容量的最大值设置
33. agent\_lxw1234.channels.fileChannel.transactionCapacity = 200
34.

## TaildirSource的配置说明（带\*的为必须配置） ##

**\*\*channels\*\***

**\*\*type\*\***

**\*\*filegroups\*\***     指定filegroups，可以有多个，以空格分隔；（TailSource可以同时监控tail多个目录中的文件）

**\*\*filegroups.<filegroupName>\*\***     配置每个filegroup的文件绝对路径，文件名可以用正则表达式匹配

positionFile    配置检查点文件的路径，检查点文件会以json格式保存已经tail文件的位置

## 启动Agent ##

cd $FLUME\_HOME/conf/

flume-ng agent -n agent\_lxw1234 –conf . -f agent\_lxw1234\_conf.properties

## 运行结果 ##

启动之后，在sink所指的/tmp/flumefiles目录下，生成了一个大小为0的目标文件，命令为时间戳-1，如：

![flume tail file][]

接着往监控的目录中生成log\_20151015\_10.log的文件：

![flume tail file][flume tail file 1]

此时，在上面tail –f目标文件的控制台中，已经可以看到写入的内容了：

![flume tail file][flume tail file 2]

再模拟生成一个新的文件（log\_20151015\_11.log）：

![flume tail file][flume tail file 3]

同样，目标文件中也正常写入：

![flume tail file][flume tail file 4]

如果在监控的目录/tmp/lxw1234-flume/中，产生和所配置的文件名正则表达式不匹配的文件，则不会被tail。

另外，如果将所监控目录/tmp/lxw1234-flume/中已经过期的文件移除，也不会影响agent的运行。

## 检查点文件positionFile ##

看一下该文件的内容：![flume tail file][flume tail file 5]

该文件中记录了所监控的每个文件的当前位置，如图中红圈圈出的pos的值，因为两个文件都已经读到了最后，因此每个pos的值就是该文件的大小。

TailSource使用了RandomAccessFile来根据positionFile中保存的文件位置来读取文件的，在agent重启之后，亦会先从positionFile中找到上次读取的文件位置，保证内容不会重复发送。

[Link 1]: http://img.lxw1234.com/flume-taildirfile-source.zip
[flume tail file]: /images/20220504/6808cbadf6f5491b93f919385fdd2967.png
[flume tail file 1]: /images/20220504/d55e87036ab14d1e8ef5f8ae8822114b.png
[flume tail file 2]: /images/20220504/e62ede85d19e4272a82df969ce61058e.png
[flume tail file 3]: /images/20220504/b50653281ff347a2a580c74c42e728fa.png
[flume tail file 4]: /images/20220504/4918168ae5ed4001b0666df84b3809da.png
[flume tail file 5]: /images/20220504/8b695dce9334469099b5cadff43641cb.png