小土刀

Rsyslog 连接 Kafka 指南

从去耦合和安全性考虑,最近在负责把公司内各个业务的日志统一接入到 Kafka 再进行下一步的处理,除了常见的 ELK 方案,我还尝试了系统自带的 rsyslog,以下是具体的配置过程。


更新记录

  • 2016.08.17: 初稿
  • 2016.11.19: 更新通天塔之日志分析平台系列文章链接

通天塔之日志分析平台系列文章

Nginx 配置

因为 Kafka 集群的通讯是走内网 ip,而外网访问的端口因为安全考虑只开了少数几个(这里是 8080),所以我们用 Nginx 反向代理来连通内外网

upstream mq_pool{
server ip1:13647 weight=1 max_fails=3 fail_timeout=30s;
server localhost:13647 weight=1 max_fails=3 fail_timeout=30s;
}
server{
listen 8080;
allow all;
proxy_pass mq_pool;
proxy_connect_timeout 24h;
proxy_timeout 24h;
}

这个配置的意思大概是把所有 8080 端口的消息转发到 mq_pool 的两台机器上(负载均衡),其他的就是常规配置。

Kafka server 配置

用 Nginx 反向代理之后,可以认为完全是内网环境,所以 Kafka 设置中的advertised.listeners 是唯一能设定外网访问的地方,所以要设置为外网可以访问的地址(也就是 Nginx 中配置的地址和端口)

然后需要注意的是修改 config 文件夹中的 producer 和 consumer 的配置,把集群的信息统一配置好

如果在测试 producer/consumer 的时候出现 WARN Error while fetching metadata with correlation id 错误,一般重新开启 producer/ consumer 即可(可能需要多试几次)

Python 测试

为了检测外网访问,我们可以用 python 进行测试,先安装一个 Kafka 客户端

pip install kafka-python

然后在 python 中执行

from kafka import KafkaProducer
producer = KafkaProducer(bootstrap_servers='url:port')
producer.send('test', 'hello from python')

如果一切正常,应该可以在 Kafka 的 consumer 中见到这条消息。

Rsyslog 配置

一般来说默认的 rsyslog 版本较低,如果想要连接 kafka,我们需要先升级到最新版本。这里可以用 rsyslogd -N1 来查看目前机器上的版本。

执行下面的命令升级

sudo add-apt-repository ppa:adiscon/v8-stable
sudo apt-get update
sudo apt-get install rsyslog
# 安装 kafka 插件
sudo apt-get install rsyslog-kafka rsyslog-imptcp

如果一切正常,我们就可以来配置 rsyslog 了,打开 /etc/rsyslog.d/ 文件夹,创建一个配置文件

module(load="omkafka")
module(load="imfile")
ruleset(name="kafka") {
action (
type="omkafka"
topic="test"
broker="broker.ip:port"
template="json"
)
}
input(type="imfile" Tag="wdxtub" File="/data/log/logstash_production.log" Ruleset="kafka")

然后我们重启服务 sudo service rsyslog restart 应该就可以在 kafka 自带的 consumer 中看到日志了。

bin/kafka-console-consumer.sh --zookeeper bi03:2181,bi02:2181 --from-beginning --topic test

至此,配置完成。

参考链接

您的支持是对我创作最大的鼓励!

热评文章