iLogtail是阿里云日志服务（SLS）团队自研的可观测数据采集Agent，拥有的轻量级、高性能、自动化配置等诸多生产级别特性，可以署于物理机、虚拟机、Kubernetes等多种环境中来采集遥测数据。iLogtail在阿里云上服务了数万家客户主机和容器的可观测性采集工作，在阿里巴巴集团的核心产品线，如淘宝、天猫、支付宝、菜鸟、高德地图等也是默认的日志、监控、Trace等多种可观测数据的采集工具。目前iLogtail已有千万级的安装量，每天采集数十PB的可观测数据，广泛应用于线上监控、问题分析/定位、运营分析、安全分析等多种场景，在实战中验证了其强大的性能和稳定性。
在当今云原生的时代，我们坚信开源才是iLogtail最优的发展策略，也是释放其最大价值的方法。因此，我们决定将iLogtail开源，期望同众多开发者一起将iLogtail打造成世界一流的可观测数据采集器。
背景
日志作为可观测性建设中的重要一环，可以记录详细的访问请求以及错误信息，在业务分析、问题定位等方面往往会发挥很大的作用。一般开发场景下，当需要进行日志分析时，往往是直接在日志文件中进行grep搜索对应的关键字；但在大规模分布式生产环境下，此方法效率低下，常见解决思路是建立集中式日志收集系统，将所有节点上的日志统一收集、管理、分析。目前市面上比较主流的开源方案是基于ELK构建一套日志采集系统。
该架构中，Filebeat作为日志源的采集Agent部署在业务集群上进行原始日志采集，并采集到的日志发送到消息队列Kafka集群。之后，由Logstash从Kafka消费数据，并经过过滤、处理后，将标准化后的日志写入Elasticsearch集群进行存储。最后，由Kibana呈现给用户查询。
该架构两个关键设计考虑：
- 引入
Kafka。
- 有助于上下游解耦。采集端可以专注于日志采集，数据分析段专注于从
Kafka消费数据，Kafka也支持重复消费，可以有效降低日志重复采集对端的影响。 - 当业务量增长时，容易出现日志量突增的情况，直接写入ES很可能出现性能下降，甚至有ES集群宕机的风险。引入
Kafka能很好的解决流量峰值的问题，削峰填谷，可以保证写入ES集群流量的均衡。 Kafka作为消息队列具有持久化的能力，数据不会丢失，吞度量大，能够很好的防止数据丢失。
- 使用
Filebeat作为日志源的采集Agent。
Filebeat是一个轻量级的日志传输工具，作为采集端Agent，可以有效弥补Logstash的缺点（例如，依赖java、数据量大时资源消耗过多）。
虽然引入了Filebeat作为日志源的采集Agent，可以有效提升端上的采集效率及资源占用情况，但是超大流量场景下Filebeat依然会有些力不从心。此外，容器场景下Filebeat支持也不是很好。而iLogtail作为阿里云日志服务（SLS）团队自研的可观测数据采集Agent，在日志采集性能及K8s支持上都有不错的体验，因此我们可以将iLogtail引入到日志采集系统的架构中。由于iLogtail有明显的性能优势，且具有极强的数据处理能力，因此可以将Logstash所承载的日志处理前置到iLogtail来进行，这样可以有效降低用户的存储成本。
此外，如果iLogtail跟阿里云日志服务（SLS）有天然的集成优势，当使用SLS作为后端存储系统时，可以直接写入，不需要额外的再引入消息队列。
本文将会详细介绍如何使用iLogtail社区版将日志采集到Kafka中，从而帮助使用者构建日志采集系统。
场景
采集/root/bin/input_data/access.log、/root/bin/input_data/error.log，并将采集到的日志写入本地部署的kafka中。
其中，access.log需要正则解析；error.log为单行文本打印。
前提条件
- 安装
# 下载、解压
$ wget https://dlcdn.apache.org/kafka/3.2.0/kafka_2.13-3.2.0.tgz
$ tar -xzf kafka_2.13-3.2.0.tgz
$ cd kafka_2.13-3.2.0
# 启动
# Start the ZooKeeper service
# Note: Soon, ZooKeeper will no longer be required by Apache Kafka.
$ nohup bin/zookeeper-server-start.sh config/zookeeper.properties &
# Start the Kafka broker service
$ nohup bin/kafka-server-start.sh config/server.properties &
- 分别创建两个topic，用于存储access-log、error-log。
bin/kafka-topics.sh --create --topic access-log --bootstrap-server localhost:9092
bin/kafka-topics.sh --create --topic error-log --bootstrap-server localhost:9092
- 更多部署说明，详见链接。
安装ilogtail
- 下载
tar -xzvf ilogtail-1.1.0.linux-amd64.tar.gz
$ cd ilogtail-1.1.0
$ ll
drwxrwxr-x 5 505 505 4096 7月 10 18:00 example_config
-rwxr-xr-x 1 505 505 84242040 7月 11 00:00 ilogtail
-rwxr-xr-x 1 505 505 16400 7月 11 00:00 libPluginAdapter.so
-rw-r--r-- 1 505 505 115963144 7月 11 00:00 libPluginBase.so
-rw-rw-r-- 1 505 505 11356 7月 11 00:00 LICENSE
-rw-rw-r-- 1 505 505 5810 7月 11 00:00 README-cn.md
-rw-rw-r-- 1 505 505 4834 7月 11 00:00 README.md
drwxr-xr-x 2 root root 4096 7月 12 09:55 user_yaml_config.d
- 采集配置
在user_yaml_config.d创建针对access_log、error_log分别创建两个采集配置，两个采集配置分别将日志采集到Kafka不同的Topic中。
# 访问日志采集配置
$ cat user_yaml_config.d/access_log.yaml
enable: true
inputs:
- Type: file_log
LogPath: /root/bin/input_data/
FilePattern: access.log
processors:
- Type: processor_regex
SourceKey: content
Regex: ([\d\.]+) \S+ \S+ \[(\S+) \S+\] \"(\w+) ([^\\"]*)\" ([\d\.]+) (\d+) (\d+) (\d+|-) \"([^\\"]*)\" \"([^\\"]*)\"
Keys:
- ip
- time
- method
- url
- request_time
- request_length
- status
- length
- ref_url
- browser
flushers:
- Type: flusher_kafka
Brokers:
- localhost:9092
Topic: access-log
# 错误日志采集配置
$ cat user_yaml_config.d/error_log.yaml
enable: true
inputs:
- Type: file_log
LogPath: /root/bin/input_data/
FilePattern: error.log
flushers:
- Type: flusher_kafka
Brokers:
- localhost:9092
Topic: error-log
$ tree user_yaml_config.d/
user_yaml_config.d/
├── access_log.yaml
└── error_log.yaml
- 启动
$ nohup ./ilogtail > stdout.log 2> stderr.log &
验证
- 访问日志验证
# 终端1: 启动kafka-console-consumer，消费access-log
$ bin/kafka-console-consumer.sh --topic access-log --from-beginning --bootstrap-server localhost:9092
# 终端2: 写入访问日志
$ echo '127.0.0.1 - - [10/Aug/2017:14:57:51 +0800] "POST /PutData?Category=YunOsAccountOpLog HTTP/1.1" 0.024 18204 200 37 "-" "aliyun-sdk-java"' >> /root/bin/input_data/access.log
# 终端1: 消费到写入的访问日志，说明流程正常。
{"Time":1657592120,"Contents":[{"Key":"__tag__:__path__","Value":"/root/bin/input_data/access.log"},{"Key":"ip","Value":"127.0.0.1"},{"Key":"time","Value":"10/Aug/2017:14:57:51"},{"Key":"method","Value":"POST"},{"Key":"url","Value":"/PutData?Category=YunOsAccountOpLog HTTP/1.1"},{"Key":"request_time","Value":"0.024"},{"Key":"request_length","Value":"18204"},{"Key":"status","Value":"200"},{"Key":"length","Value":"37"},{"Key":"ref_url","Value":"-"},{"Key":"browser","Value":"aliyun-sdk-java"}]}
- 错误日志验证
# 终端1: 启动kafka-console-consumer，消费error-log
$ bin/kafka-console-consumer.sh --topic error-log --from-beginning --bootstrap-server localhost:9092
# 终端2: 写入错误日志
$ echo -e '2022-07-12 10:00:00 ERROR This is a error!\n2022-07-12 10:00:00 ERROR This is a new error!' >> /root/bin/input_data/error.log
# 终端1: 消费到写入的错误日志，说明流程正常。
{"Time":1657591799,"Contents":[{"Key":"__tag__:__path__","Value":"/root/bin/input_data/error.log"},{"Key":"content","Value":"2022-07-12 10:00:00 ERROR This is a error!"}]}
{"Time":1657591799,"Contents":[{"Key":"__tag__:__path__","Value":"/root/bin/input_data/error.log"},{"Key":"content","Value":"2022-07-12 10:00:00 ERROR This is a new error!"}]}
总结
以上，我们介绍了使用iLogtail社区版将日志采集到Kafka的方法，大家可以与其他开源软件Kafka、ELK配合，构建出属于自己的开源日志采集系统；同样的，如果对采集的稳定性、查询的体验有更高的要求，也可以基于SLS构建商业版的可观测平台。
关于iLogtail
iLogtail作为阿里云SLS提供的可观测数据采集器，可以运行在服务器、容器、K8s、嵌入式等多种环境，支持采集数百种可观测数据（日志、监控、Trace、事件等），已经有千万级的安装量。目前，iLogtail已正式开源，欢迎使用及参与共建。
GitHub:https://github.com/alibaba/ilogtail
社区版文档：https://ilogtail.gitbook.io/ilogtail-docs/about/readme
企业版官网：https://help.aliyun.com/document_detail/65018.html
钉钉群：iLogtail社区
