码迷,mamicode.com
首页 > 编程语言 > 详细

ELK+zookeeper+kafka收集springcould日志配置文档

时间:2019-01-10 19:30:48      阅读:232      评论:0      收藏:0      [点我收藏+]

标签:xxx   input   auto   ISE   auth   客户   LTP   .com   .sh   

技术分享图片
####
####
####

技术分享图片

本文介绍使用ELK(elasticsearch、logstash、kibana) + kafka来搭建一个日志系统。主要演示使用spring aop进行日志收集,然后通过kafka将日志发送给logstash,logstash再将日志写入elasticsearch,这样elasticsearch就有了日志数据了,最后,则使用kibana将存放在elasticsearch中的日志数据显示出来,并且可以做实时的数据图表分析等等。

最开始我些项目的时候,都习惯用log4j来把日志写到log文件中,后来项目有了高可用的要求,我们就进行了分布式部署web,这样我们还是用log4j这样的方式来记录log的话,那么就有N台机子的N个log目录,ELK正常使用时需要在每台机器上都安装logstash等收集日志的客户端,这个时候查找log起来非常麻烦,不知道问题用户出错log是写在哪一台服务器上的,后来,想到一个办法,干脆把log直接写到数据库中去,这样做,虽然解决了查找异常信息便利性的问题了,但存在两个缺陷:

1,log记录好多,表不够用啊,又得分库分表了,

2,连接db,如果是数据库异常,那边log就丢失了,那么为了解决log丢失的问题,那么还得先将log写在本地,然后等db连通了后,再将log同步到db,这样的处理办法,感觉是越搞越复杂。

现在ELK的做法
好在现在有了ELK这样的方案,可以解决以上存在的烦恼,首先是,使用elasticsearch来存储日志信息,对一般系统来说可以理解为可以存储无限条数据,因为elasticsearch有良好的扩展性,然后是有一个logstash,可以把理解为数据接口,为elasticsearch对接外面过来的log数据,它对接的渠道,有kafka,有log文件,有redis等等,足够兼容N多log形式,最后还有一个部分就是kibana,它主要用来做数据展现,log那么多数据都存放在elasticsearch中,我们得看看log是什么样子的吧,这个kibana就是为了让我们看log数据的,但还有一个更重要的功能是,可以编辑N种图表形式,什么柱状图,折线图等等,来对log数据进行直观的展现。

技术分享图片

ELK职能分工

logstash做日志对接,接受应用系统的log,然后将其写入到elasticsearch中,logstash可以支持N种log渠道,kafka渠道写进来的、和log目录对接的方式、也可以对reids中的log数据进行监控读取,等等。

elasticsearch存储日志数据,方便的扩展特效,可以存储足够多的日志数据。

kibana则是对存放在elasticsearch中的log数据进行:数据展现、报表展现,并且是实时的。
技术分享图片

此次安装的软件:es,kibana,logstash,kafka,zookeeper
ELK版本最好选用5以上的版本,es需要用普通用户起服务还需要注意系统的最大打开文件数等参数,索引的应用基本都是解压即可。
修改系统参数

vi /etc/security/limits.conf

*              soft    nproc          65536

*              hard    nproc          65536

*              soft    nofile          65536

*              hard    nofile          65536

vi /etc/sysctl.conf

vm.max_map_count= 262144

sysctl -p 

####
####

es:
技术分享图片

vim elasticsearch.yml

node.name : elk1

path.data : /data/elk_data

path.logs : /data/elk_data

network.host : 192.168.0.181

http.port : 9200

http.cors.enabled: true

http.cors.allow-origin: "*"
##
由于head插件本质上还是一个nodejs的工程,因此需要安装node,使用npm来安装依赖的包。(npm可以理解为maven)

下载node-v6.10.3-linux-x64.tar.gz ,nodejs下载地址

把“node-v6.10.3-linux-x64.tar.gz”拷贝到centos上,本示例目录为:/opt/es/
技术分享图片

cd /opt/es/

tar -xzvf node-v6.10.3-linux-x64.tar.gz

ln -s /opt/es/node-v6.10.3-linux-x64/bin/node /usr/local/bin/node

ln -s /opt/es/node-v6.10.3-linux-x64/bin/npm /usr/local/bin/npm

cd /opt/es/node-v6.10.3-linux-x64/bin/

测试是否安装成功

npm -v

#####
#####

kibana:

技术分享图片

vim kibana.yml

server.port: 5601
server.host: "192.168.0.181"
elasticsearch.url: "http://192.168.0.181:9200"
kibana.index: ".kibana"

#####
#####
技术分享图片

kafka:

技术分享图片

vim server.properties

broker.id=0
#port=9092
#host.name=192.168.0.181
listeners=PLAINTEXT://192.168.0.181:9092
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/data/logs/kafka-logs
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=192.168.0.181:2181
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0
#advertised.host.name=192.168.0.181
#advertised.port=9092
advertised.listeners=PLAINTEXT://192.168.0.181:9092
replica.fetch.max.bytes=4194304
message.max.bytes=4000000
compression.codec=snappy
max.partition.fetch.bytes=4194304

#
#
添加topic

./kafka-topics.sh --create --topic app-log --replication-factor 1 --partitions 1 --zookeeper localhost:2181
./kafka-topics.sh --list --zookeeper localhost:2181

#
./kafka-console-producer.sh --broker-list 192.168.0.181:9092 --topic app-log
./kafka-console-consumer.sh --zookeeper 192.168.0.181:2181 --topic app-log --from-beginning //查看输出

技术分享图片

####
####

logstash:

添加配置文件,启动logstash时-f指定配置文件

[root@elk-181 config]# cat kafka.conf 
input {
    kafka {
        bootstrap_servers => "192.168.0.181:9092"
        group_id => "app-log"
        topics => ["app-log"]
        consumer_threads => 5  
        decorate_events => false 
        auto_offset_reset => "earliest" #latest最新的;earliest最开始的
    }
}

filter {
        json {
            source => "message"
        }
                if [severity] == "DEBUG" {
               drop {}
               }
}

output {
    elasticsearch {
    action => "index"          #The operation on ES
        hosts  => "192.168.0.181:9200"     #ElasticSearch host, can be array.
#       index  => "applog"         #The index to write data to.  
        index => "applog-%{+YYYY.MM.dd}"
    }
  }

####
####

springcould日志配置

###
技术分享图片

###

技术分享图片

log:
  brokerList:
    url: 192.168.0.181:9092
    topic: app-log
  console :
    level : DEBUG

#
#

#
#
<appender name="KAFKA" class="com.skong.core.logs.KafkaAppender">
<filter class="ch.qos.logback.classic.filter.ThresholdFilter">

${logLevel} # # ![](https://s1.51cto.com/images/blog/201901/10/f03c34d45cc43564d98bc2dcf7a6019a.png?x-oss-process=image/watermark,size_16,text_QDUxQ1RP5Y2a5a6i,color_FFFFFF,t_100,g_se,x_10,y_10,shadow_90,type_ZmFuZ3poZW5naGVpdGk=) # # # ![](https://s1.51cto.com/images/blog/201901/10/c0e5bcced0a454b2900a9f508ff9e77e.jpg?x-oss-process=image/watermark,size_16,text_QDUxQ1RP5Y2a5a6i,color_FFFFFF,t_100,g_se,x_10,y_10,shadow_90,type_ZmFuZ3poZW5naGVpdGk=) 此脚本为定时删除索引和重启 ``` [root@elk-181 ~]# cat cleanesdata.sh #!/bin/bash #author:silence_ltx #date:20180809 PATH=/data/xxxx-node/bin:/xxxx-java/bin:/data/elasticsearch/bin:/data/kibana/bin:/usr/local/python/bin:/data/xxxx-node/bin:/data/xxxx-node/bin:/data/xxxx-java/bin:/data/elasticsearch/bin:/data/kibana/bin:/data/xxxx-java/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/root/bin:/root/bin:/root/bin date=`date +%Y.%m.%d -d "2 days ago"` echo $date /usr/bin/curl -XDELETE "192.168.0.181:9200/applog-${date}" ps -ef| grep logstash | grep -v grep | awk ‘{print $2}‘ | xargs kill -9 netstat -nltp | grep 9092 | awk ‘{print $NF}‘ | awk -F "/" ‘{print $1 }‘| xargs kill -9 sleep 30 find /data/logs/kafka-logs/app-log-0/ -mtime +2 -exec rm -rf {} \; cd /data/kafka/bin sh ./kafka-server-start.sh ../config/server.properties > /data/kafka/bin/kafka.log & sleep 30 cd /data/logstash/bin sh ./logstash -f ../config/kafka.conf > /data/logstash/bin/logstash.log & ``` ![](https://s1.51cto.com/images/blog/201901/10/32c6a3db29380f7498ca9f11a411db03.jpg?x-oss-process=image/watermark,size_16,text_QDUxQ1RP5Y2a5a6i,color_FFFFFF,t_100,g_se,x_10,y_10,shadow_90,type_ZmFuZ3poZW5naGVpdGk=) # # # ![](https://s1.51cto.com/images/blog/201901/10/8d40ae54dd476149390d0fef36c8322c.jpg?x-oss-process=image/watermark,size_16,text_QDUxQ1RP5Y2a5a6i,color_FFFFFF,t_100,g_se,x_10,y_10,shadow_90,type_ZmFuZ3poZW5naGVpdGk=)

ELK+zookeeper+kafka收集springcould日志配置文档

标签:xxx   input   auto   ISE   auth   客户   LTP   .com   .sh   

原文地址:http://blog.51cto.com/13386520/2341262

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