rsyslog对接kafka

759876uyt

对于日志收集统计分析方案现在主流的有ELK redis kafka flume +大数据平台，ELK我们以后再介绍，这里先简单介绍下rsyslog对接kafkakafka中的术语

• Broker
  
  Kafka集群包含一个或多个服务器，这种服务器被称为broker
• Topic
  
  每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic。（物理上不同Topic的消息分开存储，逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处）
• Partition
  
  Parition是物理上的概念，每个Topic包含一个或多个Partition.
• Producer
  
  负责发布消息到Kafka broker
• Consumer
  
  消息消费者，向Kafka broker读取消息的客户端。
• Consumer Group
  
  每个Consumer属于一个特定的Consumer Group（可为每个Consumer指定group name，若不指定group name则属于默认的group）
  
  

kafka架构


下载kafka_2.10-0.10.1.0.tgz  解压，这里我们搞个单机测试，不做kafka集群
[iyunv@k8snode1 kafka_2.10-0.10.1.0]# ls
bin  config  libs  LICENSE  logs  NOTICE  site-docs

1 2 3 4 5 6 7 8 9 10 11 12 13 14

./bin/zookeeper-server-start.sh zookeeper.properties  先启动zookeeper 新建一个名字叫test_nginx的topic： bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test_nginx 查看当前存在的topic： bin/kafka-topics.sh --list --zookeeper localhost:2181 [iyunv@k8snode1 kafka_2.10-0.10.1.0]# ./bin/kafka-server-start.sh config/server.properties > /dev/null 2>&1 &   启动kafka [iyunv@k8snode1 kafka_2.10-0.10.1.0]# bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test_nginx  创建生产者 [iyunv@k8snode1 kafka_2.10-0.10.1.0]# bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test_nginx --from-beginning   创建消费者

查看启动情况


在生产者输入 命令测试，

在消费者端观察看是否有输出


这说明我们的kafka环境构建好了

通过对 rsyslog官方文档 查看,得知 rsyslog对 kafka的支持是 v8.7.0版本后才提供的支持.通过 ChangeLog 也可以看出 V8.X的版本变化.

http://www.rsyslog.com/doc/v8-st ... odules/omkafka.html

最新V8稳定版已经提供RPM包的Rsyslog-kafka插件了,直接yum安装即可,添加yum源：
[rsyslog_v8]
name=Adiscon CentOS-$releasever - local packages for $basearch
baseurl=http://rpms.adiscon.com/v8-stable/epel-$releasever/$basearch
enabled=1
gpgcheck=0
gpgkey=http://rpms.adiscon.com/RPM-GPG-KEY-Adiscon
protect=1


yum install -y rsyslog rsyslog-kafka

[iyunv@k8snode1 yum.repos.d]# rpm -ql rsyslog-kafka
/lib64/rsyslog/omkafka.so   #rsyslog  连接  kafka的插件

[iyunv@k8snode1 rsyslog.d]# rsyslogd -N1
rsyslogd: version 8.24.0, config validation run (level 1), master config /etc/rsyslog.conf

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68

[iyunv@k8snode1 rsyslog.d]# pwd /etc/rsyslog.d [iyunv@k8snode1 rsyslog.d]# cat nginx_test.conf #加载omkafka和imfile模块 module(load="omkafka") module(load="imuxsock") # push to kafka action(type="omkafka" topic="test_nginx" broker="localhost:9092") # 定义消息来源及设置相关的action这是直接 这是直接收集rsyslog日志，假如是应用程序产生的日志可以先写入文件，rsyslog在从文件读取日志 [iyunv@k8snode1 rsyslog.d]# cat nginx_kafka.conf #加载omkafka和imfile模块 module(load="omkafka") module(load="imfile") # nginx template template(name="nginxAccessTemplate" type="string" string="%hostname%<-+>%syslogtag%<-+>%msg%\n") # ruleset ruleset(name="nginx-kafka") {     #日志转发kafka     action (         type="omkafka"         template="nginxAccessTemplate"         confParam=["compression.codec=snappy", "queue.buffering.max.messages=400000"]         partitions.number="4"         topic="test_nginx"         broker="localhost:9092"         queue.spoolDirectory="/tmp"         queue.filename="test_nginx_kafka"         queue.size="360000"         queue.maxdiskspace="2G"         queue.highwatermark="216000"         queue.discardmark="350000"         queue.type="LinkedList"         queue.dequeuebatchsize="4096"         queue.timeoutenqueue="0"         queue.maxfilesize="10M"         queue.saveonshutdown="on"         queue.workerThreads="4"     ) } # 定义消息来源及设置相关的action input(type="imfile" Tag="nginx,aws" File="/usr/local/nginx/logs/access.log" Ruleset="nginx-kafka") 重启rsyslog，并往rsylog写入数据测试 [iyunv@k8snode1 rsyslog.d]# logger 'hello world123' [iyunv@k8snode1 rsyslog.d]# logger 'hello world1' [iyunv@k8snode1 rsyslog.d]# logger 'hello world2' [iyunv@k8snode1 rsyslog.d]# logger 'hello world3' 观察kafka消费者，可以看到已经有输出了，说明rsyslog对接kafka成功 2017-02-13T10:07:56.711023+08:00 k8snode1 root: hello world123 2017-02-13T10:08:05.687581+08:00 k8snode1 root: hello world1 2017-02-13T10:08:07.877318+08:00 k8snode1 root: hello world2 2017-02-13T10:08:09.757482+08:00 k8snode1 root: hello world3