码迷,mamicode.com
首页 > 其他好文 > 详细

rsyslog对接kafka

时间:2017-02-13 14:20:27      阅读:876      评论:0      收藏:0      [点我收藏+]

标签:elk kafka rsyslog flume

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

kafka中的术语

  • 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集群

[root@k8snode1 kafka_2.10-0.10.1.0]# ls

bin  config  libs  LICENSE  logs  NOTICE  site-docs



./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

[root@k8snode1 kafka_2.10-0.10.1.0]# ./bin/kafka-server-start.sh config/server.properties > /dev/null 2>&1 &   启动kafka


[root@k8snode1 kafka_2.10-0.10.1.0]# bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test_nginx  创建生产者

[root@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-stable/configuration/modules/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


[root@k8snode1 yum.repos.d]# rpm -ql rsyslog-kafka

/lib64/rsyslog/omkafka.so   #rsyslog  连接  kafka的插件


[root@k8snode1 rsyslog.d]# rsyslogd -N1

rsyslogd: version 8.24.0, config validation run (level 1), master config /etc/rsyslog.conf

[root@k8snode1 rsyslog.d]# pwd
/etc/rsyslog.d
[root@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在从文件读取日志

[root@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写入数据测试
[root@k8snode1 rsyslog.d]# logger ‘hello world123‘
[root@k8snode1 rsyslog.d]# logger ‘hello world1‘
[root@k8snode1 rsyslog.d]# logger ‘hello world2‘
[root@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

本文出自 “厚德载物” 博客,谢绝转载!

rsyslog对接kafka

标签:elk kafka rsyslog flume

原文地址:http://huaxin.blog.51cto.com/903026/1897188

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!