rsyslog & kafka 流量日志
Posted 淡季的风
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了rsyslog & kafka 流量日志相关的知识，希望对你有一定的参考价值。
1、rsyslog安装
在线安装: https://www.rsyslog.com/rhelcentos-rpms/
离线安装:
1）版本要求:
操作系统: Centos 7.2
应用版本: 8.34.0
2）依赖:
(一般来说安装最新版本的依赖就可以)
必须：
Gcc
Libestr 0.1.10
Libfastjson 0.99.8
非必须（某些插件要求）:
3）依赖安装：
a) Libestr
下载libestr最新版安装包， 解压到某个目录， 进入该目录；
执行： ./configure --libdir=/usr/lib64 --includedir=/usr/include
执行: make && make install
查看/usr/lib64目录下是否出现libestr.so等链接库，
查看/usr/lib64/pkgconfig目录下是否出现libestr.pc
b) Libfastjson
下载libfastjson最新版安装包， 解压到某个目录， 进入该目录
执行: ./configure --libdir=/usr/lib64 --includedir=/usr/include
执行：make && make install
查看/usr/lib64目录下是否出现libfastjson.so等链接库，
查看/usr/lib64/pkgconfig目录下是否出现libfastjson.pc
4) rsyslog 安装:
下载rsyslog 8.34.0的安装包。
http://rpms.adiscon.com/v8-stable/epel-7/x86_64/RPMS/rsyslog-8.34.0-1.el7.centos.x86_64.rpm
rpm -i rsyslog-8.34.0-2.el7.centos.x86_64.rpm
输入rsyslogd –version 查看版本号， 是否安装成功
5) redis插件安装
rsyslog的redis插件依赖于hiredis, 下载hiredis 0.13.0 rpm包
rpm –i hiredis-0.13.3-7.fc28.x86_64.rpm
下载rsyslog-hiredis插件包
rpm -i --nodeps rsyslog-hiredis-8.34.0-1.fc29.x86_64.rpm
（--nodeps 否则会失败）
查看/usr/lib64/rsyslog/omhiredis.so 是否存在
6) kafka 插件安装
下载liblz4 rpm包安装
rpm –i --nodeps lz4-1.7.3-1.el7.x86_64.rpm
下载rsyslog-kafka插件包
rpm -i –nodeps rsyslog-kafka-8.34.0-2.el7.centos.x86_64.rpm
查看/usr/lib64/rsyslog/omkafka.so是否存在
2、Kafka安装
1) 版本要求
Kafka 1.1.0
2) 依赖
Java 1.8
Zookeeper
3) 依赖安装
Java
Zookeeper(kafka自带，但zookeeper一般至少3个节点)
4） kafka安装
略
5）需要注意问题
Zookeeper和kafka一般都是集群方式部署， 流量采集服务器单台怎么部署集群，磁盘问题。
3、Rsyslog插件
1） redis
# Provides UDP syslog reception $ModLoad imudp $UDPServerRun 514 # Provides TCP syslog reception $ModLoad imtcp $InputTCPServerRun 514 # An on-disk queue is created for this action. If the remote host is # down, messages are spooled to disk and sent when it is up again. #$ActionQueueFileName /data/rsyslog/queue/syslog # unique name prefix for spool files #$ActionQueueMaxDiskSpace 10g # 1gb space limit (use as much as possible) #$ActionQueueSaveOnShutdown on # save messages to disk on shutdown #$ActionQueueType LinkedList # run asynchronously #$ActionResumeRetryCount -1 # infinite retries if host is down # remote host is: name/ip:port, e.g. 192.168.0.1:514, port optional #*.* @@remote-host:514 # ### end of the forwarding rule ### $ModLoad omhiredis #module(load="omhiredis") action( type="omhiredis" name="push_redis" server="127.0.0.1" serverport="6379" mode="queue" key="syslog" )
2) kafka
# Provides UDP syslog reception #$ModLoad imudp #$UDPServerRun 514 # Provides TCP syslog reception #$ModLoad imtcp #$InputTCPServerRun 514 #module(load="imtcp" maxsessions="1000") #input( # type="imtcp" # address="0.0.0.0" # port="514" #) module(load="imudp" threads="2" timeRequery="8" batchSize="128") module(load="imptcp" threads="4") input(type="imudp" port="514" ruleset="write2kafka") input(type="imptcp" port="514" ruleset="write2kafka") # Where to place auxiliary files # # # Use default timestamp format #$ActionFileDefaultTemplate RSYSLOG_TraditionalFileFormat template(name="rawFormat" type="string" string="%rawmsg%") #$ActionFileDefaultTemplate rawFormat # # remote host is: name/ip:port, e.g. 192.168.0.1:514, port optional #*.* @@remote-host:514 # ### end of the forwarding rule ### #$ModLoad omkafka # module(load="omkafka") ruleset( name="write2kafka" queue.filename="syslog" queue.spoolDirectory="/data/rsyslog/queue" queue.size="10000000" queue.maxdiskspace="10G" queue.type="LinkedList" queue.saveonshutdown="on" queue.dequeueBatchSize="4096" queue.workerThreads="10" queue.workerThreadMinimumMessages="100000" ) { action( template="rawFormat" type="omkafka" topic="syslog" confParam=["compression.codec=snappy", "socket.timeout.ms=20", "socket.keepalive.enable=true"] partitions.auto = "on" resubmitonfailure = "on" ) }
4、Rsyslog 配置
rsyslog有3种不同配置方式: basic（已过时，兼容）, advance, obsolete legacy(已过时，兼容， 官方建议不要使用这种方式)。
目前rsyslog主要使用basic和advance混搭的方式进行配置， 凡是$开头的配置，都是obsolete legacy风格的配置， 大都有。
rsyslog的配置主体分为input, ruleset, action, global，module
1） input
input代表输入， 默认是从文件中获取输入, 其他还有imtcp. imudp等
可以使用input（）函数声明输入， input函数的type参数是必须的， 必须是输入源的module名，如tcp使用imtcp模块， 则type必须是imtcp，其他参数根据不同的module而定， 如tcp必须约定address（主机名）, port（监听的端口）等等。
以tcp为例， 输入配置为:
input(
type="imtcp"
address="0.0.0.0"
port="514"
)
或者
$InputTCPServerRun 514
可以有多个输入源
2） ruleset
略
3） action
action即output， 默认输出omfile, 其他还有omhiredis, omfwd, omkafka, omelasticseasch等。
可以使用action（）函数声明输出， action的type参数必须是输出插件的module名，其他参数依不同插件而定， 需要说明的是action里面可以设置队列参数，如queue.type和queue.size等，具体见队列
4) global
全局参数, 可以使用global（）函数约定，或$参数名约定。
5） module
导入模块的函数module（）， 无论input或是action等， 使用的插件都必须通过module方法导入，module函数的load必须， 必须是插件的名称,如imfile, omfile等。其他可选参数依据不同的module而定， 可以性能调优。
也可通过$ModLoad module_name的方式导入模块。
6） include
等同于$IncludeConfig
7） queue
rsyslog对列分为direct, 磁盘， 内存3种模式， direct和磁盘模式都影响性能，内存模式分为FixedArray和LinkedList， 一般情况下使用LinkedList比较好。
也可以启用DA模式，内存队列和磁盘队列结合使用， 默认情况下，优先使用内存队列，当内存中队列已满时，开启磁盘队列。即可以保证高性能， 又可以防止数据丢失。
重要参数:
queue.size 队列的最大大小，不能设太小，影响性能
queue.type 可选值：FixedArray / LinkedList / Direct / Disk
queue.saveonshutdown on/off
queue.workerthreads 工作线程数，官方建议1
queue.filename 存储队列的文件名称， 仅对磁盘队列有用， 需要注意的是只能是文件名，不能包含目录
queue.spoolDirectory 存储队列的工作目录， 仅对磁盘队列有用
queue.maxdiskspace 存储队列的最大大小，仅对磁盘队列有用
queue.highwatermark 仅对DA模式有用， 队列大小达到该水平，开启磁盘辅助队列
queue.lowwatermark仅对DA模式有用，开启磁盘辅助队列后， 如果队列大小降到该水平， 关闭磁盘辅助队列，回复内存模式
5、Rsyslog 运行
1) systemctl
systemctl start|status|stop rsyslog 读取/etc/rsyslog.conf 及 /etc/rsyslogd.d/*.conf
systemctl status –l rsyslog 查看rsyslog运行情况， 可以看到报错信息
2) 命令行
rsyslogd –i pid文件 –f conf配置文件
可以通过 rsyslogd –N4 查看rsyslog配置是否正确
可以通过 rsyslogd –dn 进行debug
以上是关于rsyslog & kafka 流量日志的主要内容，如果未能解决你的问题，请参考以下文章