前言
Fluentd是一个实时开源的数据收集器，基于CRuby实现，td-agent是其商业化版本，由Treasure Data公司维护。本文将介绍如何使Fluentd能够读写OSS。
安装
首先下载并安装td-agent，笔者使用的是td-agent-3.3.0-1.el7.x86_64.rpm，使用rpm命令安装:
[root@apache ~]# rpm -ivh td-agent-3.3.0-1.el7.x86_64.rpm然后，需要安装Fluentd的OSS plugin:
[root@apache ~]# /usr/sbin/td-agent-gem install fluent-plugin-aliyun-oss这里请注意，因为我们使用的是td-agent，安装Fluentd plugin时需要使用td-agent的td-agent-gem(/usr/sbin/td-agent-gem)。原因是td-agent有自己的Ruby，你需要将plugin安装到它的Ruby里面，而不是其他的Ruby，否则将会找不到已经安装好的plugin。
具体可以参见Fluentd的官方文档
安装完成后，我们可以查看安装的OSS plugin:
[root@apache ~]# /usr/sbin/td-agent-gem list fluent-plugin-aliyun-oss
*** LOCAL GEMS ***
fluent-plugin-aliyun-oss (0.0.1)fluent-plugin-aliyun-oss这个plugin包含两部分：
Fluent OSS output plugin
将数据缓存在本地，达到设定的条件后，将缓存的数据(压缩后，如果设置的话)上传到OSS。
Fluent OSS input plugin
首先，OSS的bucket需要配置事件通知，这篇文章介绍了如何设置，得到MNS的Queue与Endpoint。
设置好之后，这个plugin会定时地从MNS拉取消息，从消息中获取上传的Objects，最后从OSS读取这些Objects，再发往下游。
下面将分别介绍如何配置，具体的配置参数说明参见github: https://github.com/aliyun/fluent-plugin-oss
配置(向OSS写数据)
下面是一个例子，将读到的数据，每分钟一个文件写到OSS中。其中endpoint/bucket/access_key_id/access_key_secret是必填项，其他的都是可选项。
<system>
workers 6
</system>
<match input.*>
@type oss
endpoint <OSS endpoint to connect to>
bucket <Your Bucket>
access_key_id <Your Access Key>
access_key_secret <Your Secret Key>
upload_crc_enable false
path "fluent-oss/logs"
auto_create_bucket true
key_format "%{path}/%{time_slice}/events_%{index}_%{thread_id}.%{file_extension}"
#key_format %{path}/events/ts=%{time_slice}/events_%{index}_%{thread_id}.%{file_extension}
time_slice_format %Y%m%d-%H
store_as gzip
<buffer tag,time>
@type file
path /var/log/fluent/oss/${ENV['SERVERENGINE_WORKER_ID']}
timekey 60 # 1 min partition
timekey_wait 1s
# timekey_use_utc true
flush_thread_count 1
</buffer>
<format>
@type json
</format>
</match>我们可以从OSS控制台看到效果:
配置(从OSS读数据)
下面是一个配置示例，其中endpoint/bucket/access_key_id/access_key_secret和MNS的endpoint/queue是必填项，其他的都是可选项。
<source>
@type oss
endpoint <OSS endpoint to connect to>
bucket <Your Bucket>
access_key_id <Your Access Key>
access_key_secret <Your Secret Key>
store_local false
store_as gzip
flush_batch_lines 800
flush_pause_milliseconds 1
download_crc_enable false
<mns>
endpoint <MNS endpoint to connect to, E.g.,{account-id}.mns.cn-zhangjiakou-internal.aliyuncs.com>
queue <MNS queue>
poll_interval_seconds 1
</mns>
<parse>
@type json
</parse>
</source>我们可以从log中看一下运行状态
2019-04-23 15:38:14 +0800 [info]: #5 start to poll message from MNS queue fluentd-oss
2019-04-23 15:38:14 +0800 [info]: #5 http://1305310278558820.mns.cn-zhangjiakou-internal.aliyuncs.com/queues/fluentd-oss/messages
2019-04-23 15:38:14 +0800 [info]: #5 read object fluent-oss/logs/20190423-12/events_10_70226640160100.gz, size 4389548 from OSS
2019-04-23 15:38:15 +0800 [info]: #1 http://1305310278558820.mns.cn-zhangjiakou-internal.aliyuncs.com/queues/fluentd-oss/messages?ReceiptHandle=0BC1EA4E51483D4EAC69736941044AAE-MjY5ODkgMTU1NjAwNTAwODMwNSAzNjAwMDA
2019-04-23 15:38:16 +0800 [info]: #1 start to poll message from MNS queue fluentd-oss
2019-04-23 15:38:16 +0800 [info]: #1 http://1305310278558820.mns.cn-zhangjiakou-internal.aliyuncs.com/queues/fluentd-oss/messages
2019-04-23 15:38:16 +0800 [info]: #1 read object fluent-oss/logs/20190423-09/events_50_69939581261780.gz, size 6750045 from OSS参考资料
https://github.com/aliyun/fluent-plugin-oss
https://rubygems.org/gems/fluent-plugin-aliyun-oss
https://www.fluentd.org/
https://docs.fluentd.org/v1.0/articles/quickstart
