zabbix Server 4.0监控Flume关键参数
作者：尹正杰
版权声明：原创作品，谢绝转载！否则将追究法律责任。
Flume本身提供了http, ganglia的监控服务。当然我们也可以使用JMX的方式去监控Flume，然后只要能集成JMX监控的监控系统应该都能实现简介监控Flume，有网友说，监控Flume我们需要修改其源码让他支持zabbix监控，当然这得让咱们运维人员懂Java开发才行，而且还得有一定的功底，要是改出Bug了反而麻烦。Ganglia监控起来的确方便，但我们公司的监控系统使用的是Zabbix，有的小伙伴使用的是Open Falcon，我们建议大家统一一套监控系统，当然有2套监控系统也是可以的，留作备份也是不错的。
本片博客会手把手教你使用Zabbix Server监控Flume的重要参数。其实就是利用Flume本身提供的HTTP借口，关于zabbix和flume的部署方式我这里就不再赘述了。我假设你已经将zabbix监控系统和flue-ng进程启动成功啦。接下来我们就来动手干活。
一.启用Flume自身的Http监控功能
1>.查看Flume进程的启动脚本
[root@flume112 ~]# cat /soft/flume/conf/job/flume-conf-p2p01.properties
#定义别名
agent.sources = kafkaSource
agent.channels = kafkaSource
agent.sinks = hdfsSink#绑定关系
agent.sources.kafkaSource.channels = kafkaSource
agent.sinks.hdfsSink.channel = kafkaSource#指定source源为kafka source
agent.sources.kafkaSource.type = org.apache.flume.source.kafka.KafkaSource
agent.sources.kafkaSource.kafka.bootstrap.servers = 10.1.2.114:,10.1.2.115:,10.1.2.116:,10.1.2.117:,10.1.2.118:
agent.sources.kafkaSource.topic = account-check
agent.sources.kafkaSource.kafka.consumer.group.id = -account-check
agent.sources.kafkaSource.kafka.consumer.max.partition.fetch.bytes =
agent.sources.kafkaSource.kafka.consumer.heartbeat.interval.ms =
agent.sources.kafkaSource.kafka.consumer.rebalance.timeout.ms =
agent.sources.kafkaSource.kafka.consumer.fetch.min.bytes =
agent.sources.kafkaSource.kafka.consumer.session.timeout.ms =
agent.sources.kafkaSource.kafka.consumer.request.timeout.ms =
agent.sources.kafkaSource.interceptors = i1
agent.sources.kafkaSource.interceptors.i1.userIp = true
agent.sources.kafkaSource.interceptors.i1.type = host#指定channel类型为kafka
agent.channels.kafkaSource.type = org.apache.flume.channel.kafka.KafkaChannel
agent.channels.kafkaSource.kafka.bootstrap.servers = 10.1.2.114:,10.1.2.115:,10.1.2.116:,10.1.2.117:,10.1.2.118:
agent.channels.kafkaSource.kafka.topic = channel.account-check--
agent.channels.kafkaSource.kafka.consumer.group.id = -channel.account-check--
agent.channels.kafkaSource.kafka.consumer.heartbeat.interval.ms =
agent.channels.kafkaSource.kafka.consumer.rebalance.timeout.ms =
agent.channels.kafkaSource.kafka.consumer.fetch.min.bytes =
agent.channels.kafkaSource.kafka.consumer.session.timeout.ms =
agent.channels.kafkaSource.kafka.consumer.request.timeout.ms = #指定sink的类型为hdfs
agent.sinks.hdfsSink.type = hdfs
agent.sinks.hdfsSink.hdfs.path = hdfs://hdfs-ha/user/p2p_kafka/%Y%m%d
agent.sinks.hdfsSink.hdfs.filePrefix = ---112_p2p01_%Y%m%d_%H
agent.sinks.hdfsSink.hdfs.fileSuffix = .txt
agent.sinks.hdfsSink.hdfs.useLocalTimeStamp = true
agent.sinks.hdfsSink.hdfs.writeFormat = Text
agent.sinks.hdfsSink.hdfs.fileType=DataStream
agent.sinks.hdfsSink.hdfs.rollCount =
agent.sinks.hdfsSink.hdfs.rollSize =
agent.sinks.hdfsSink.hdfs.rollInterval =
agent.sinks.hdfsSink.hdfs.batchSize =
agent.sinks.hdfsSink.hdfs.threadsPoolSize =
agent.sinks.hdfsSink.hdfs.idleTimeout =
agent.sinks.hdfsSink.hdfs.minBlockReplicas =
agent.sinks.hdfsSink.hdfs.callTimeout=
agent.sinks.hdfsSink.hdfs.request-timeout=
agent.sinks.hdfsSink.hdfs.connect-timeout=
[root@flume112 ~]#
[root@flume112 ~]# cat /soft/flume/conf/job/flume-conf-p2p01.properties
[root@flume112 ~]# cat /soft/flume/shell/start_flume_p2p01.sh
#!/bin/bash
#@author :yinzhengjie
#blog:http://www.cnblogs.com/yinzhengjie
#EMAIL:y1053419035@qq.com
#Data:Thu Oct :: CST #启动flume自身的监控参数，默认执行以下脚本
nohup flume-ng agent -c /soft/flume/conf/job/ --conf-file=/soft/flume/conf/job/flume-conf-p2p01.properties --name agent -Dflume.monitoring.type=http -Dflume.monitoring.port= -Dflume.root.logger=INFO,console >> /soft/flume/logs/flume-http-p2p01.log >& &[root@flume112 ~]#
2>.我们启动上面的Flume agent后
[root@flume112 ~]# curl 10.1.2.112:/metrics | jq
% Total % Received % Xferd Average Speed Time Time Time Current
Dload Upload Total Spent Left Speed
165k --:--:-- --:--:-- --:--:-- 177k
{
"SINK.hdfsSink": { #这是HDFS SINK监控数据。
"ConnectionCreatedCount": "", #下一个阶段（或存储系统）创建链接的数量（如HDFS创建一个文件）。
"ConnectionClosedCount": "", #下一个阶段（或存储系统）关闭链接的数量（如HDFS中关闭一个文件）。
"Type": "SINK", #很显然，这里是SINK监控项，类型为SINK
"BatchCompleteCount": "", #批量处理event的个数等于批处理大小的数量。
"BatchEmptyCount": "", #批量处理event的个数为0的数量（空的批量的数量），如果数量很大表示source写入数据的速度比sink处理数据的速度慢很多。
"EventDrainAttemptCount": "", #sink尝试写出到存储的事件总数量。
"StartTime": "", #channel启动时的毫秒值时间。
"EventDrainSuccessCount": "", #sink成功写出到存储的事件总数量。
"BatchUnderflowCount": "", #批量处理event的个数小于批处理大小的数量（比sink配置使用的最大批量尺寸更小的批量的数量），如果该值很高也表示sink比source更快。
"StopTime": "", #channel停止时的毫秒值时间，为0表示一直在运行。
"ConnectionFailedCount": "" #下一个阶段（或存储系统）由于错误关闭的连接数量（如HDFS上一个新创建的文件由于超市而关闭）。
},
"CHANNEL.kafkaSource": { #这是KAFKA SOURCE监控数据。
"KafkaEventGetTimer": "", #KAFKA事件获取计数器。
"ChannelSize": "", #目前channel中事件的总数量，目前仅支持File Channel，Memory channel的统计数据。我这里使用的是Kafka channel。
"EventTakeAttemptCount": "", #sink尝试从channel拉取事件的总次数。这不意味着每次时间都被返回，因为sink拉取的时候channel可能没有任何数据。
"StartTime": "", #channel启动时的毫秒值时间。
"KafkaCommitTimer": "", #KAFKA提交计数器。
"ChannelCapacity": "", #channel的容量，目前仅支持File Channel，Memory channel的统计数据。我这里使用的是Kafka channel。
"ChannelFillPercentage": "1.7976931348623157E308", #channel已填入的百分比。
"Type": "CHANNEL", #很显然，这里是CHANNEL监控项，类型为CHANNEL。
"EventTakeSuccessCount": "", #sink成功从channel读取事件的总数量。
"RollbackCount": "", #回滚的次数。
"KafkaEventSendTimer": "", #KAFKA事件发送计数器。
"EventPutAttemptCount": "", #Source尝试写入Channe的事件总次数。
"EventPutSuccessCount": "", #成功写入channel且提交的事件总次数。
"StopTime": "" #channel停止时的毫秒值时间，为0表示一直在运行。
},
"SOURCE.kafkaSource": {
"KafkaEventGetTimer": "", #KAFKA事件获取计数器。
"AppendBatchAcceptedCount": "", #成功提交到channel的批次的总数量。
"AppendReceivedCount": "", #每批只有一个事件的事件总数量（与RPC调用的一个append调用相等）。
"EventAcceptedCount": "", #成功写出到channel的事件总数量。
"StartTime": "", #source启动时的毫秒值时间。
"AppendBatchReceivedCount": "", #接收到事件批次的总数量。
"KafkaCommitTimer": "", #KAFKA提交计数器。
"EventReceivedCount": "", #目前为止source已经接收到的事件总数量。
"Type": "SOURCE", #很显然，这里是SOURCE监控项，类型为SOURCE。
"KafkaEmptyCount": "", #KAFKA空的批量的数量。
"AppendAcceptedCount": "", #逐条录入的次数，单独传入的事件到Channel且成功返回的事件总数量。
"OpenConnectionCount": "", #目前与客户端或sink保持连接的总数量，目前仅支持avro source展现该度量。
"StopTime": "" #source停止时的毫秒值时间，为0表示一直在运行。
}
}
[root@flume112 ~]#
温馨提示：
上图需要注意的点我已经用深颜色标识出来啦，当然，如果你还要想了解更多度量值，可参考官方文档：http://flume.apache.org/FlumeUserGuide.html#monitoring。
3>.上面是使用jq工具查看Flume自身监控的，当然，我们也可以使用sed命令将上面的内容格式化一下
[root@flume112 ~]# curl 10.1.2.112:/metrics >/dev/null |sed -e 's/\([,]\)\s*/\1\n/g' -e 's/[{}]/\n/g' -e 's/[",]//g'SINK.hdfsSink:
ConnectionCreatedCount:
ConnectionClosedCount:
Type:SINK
BatchCompleteCount:
BatchEmptyCount:
EventDrainAttemptCount:
StartTime:
EventDrainSuccessCount:
BatchUnderflowCount:
StopTime:
ConnectionFailedCount:CHANNEL.kafkaSource:
KafkaEventGetTimer:
ChannelSize:
EventTakeAttemptCount:
StartTime:
KafkaCommitTimer:
ChannelCapacity:
ChannelFillPercentage:1.7976931348623157E308
Type:CHANNEL
EventTakeSuccessCount:
RollbackCount:
KafkaEventSendTimer:
EventPutAttemptCount:
EventPutSuccessCount:
StopTime:SOURCE.kafkaSource:
KafkaEventGetTimer:
AppendBatchAcceptedCount:
AppendReceivedCount:
EventAcceptedCount:
StartTime:
AppendBatchReceivedCount:
KafkaCommitTimer:
EventReceivedCount:
Type:SOURCE
KafkaEmptyCount:
AppendAcceptedCount:
OpenConnectionCount:
StopTime:[root@flume112 ~]#
[root@flume112 ~]# curl 10.1.2.112:/metrics >/dev/null |sed -e 's/\([,]\)\s*/\1\n/g' -e 's/[{}]/\n/g' -e 's/[",]//g' | grep EventDrainSuccessCount #我们可以进一步拿到某个监控参数！
EventDrainSuccessCount:
[root@flume112 ~]#
4>.使用awk获取某个参数的值
[root@flume112 ~]# curl 10.1.2.112:/metrics >/dev/null |sed -e 's/\([,]\)\s*/\1\n/g' -e 's/[{}]/\n/g' -e 's/[",]//g' | grep EventDrainSuccessCount
EventDrainSuccessCount:
[root@flume112 ~]#
[root@flume112 ~]#
[root@flume112 ~]# curl 10.1.2.112:/metrics >/dev/null |sed -e 's/\([,]\)\s*/\1\n/g' -e 's/[{}]/\n/g' -e 's/[",]//g' | grep EventDrainSuccessCount | awk -F ':' '{print $2}'[root@flume112 ~]#
[root@flume112 ~]#
二.zabbix agent端配置
1>.查看zabbix agent的主配置文件
[root@flume112 ~]# cat /etc/zabbix/zabbix_agentd.conf | grep Include | grep -v ^#
Include=/etc/zabbix/zabbix_agentd.d/*.conf
[root@flume112 ~]#
2>.编写采集数据的脚本
[root@flume112 ~]# cat /etc/zabbix/zabbix_agentd.d/flume_monitor.sh
#!/bin/bash
#@author :yinzhengjie
#blog:http://www.cnblogs.com/yinzhengjie
#EMAIL:y1053419035@qq.com
#Data:Thu Oct :: CST FLUME_PORT=$
METRIC_VALUE=$curl 127.0.0.1:"$FLUME_PORT"/metrics >/dev/null |sed -e 's/\([,]\)\s*/\1\n/g' -e 's/[{}]/\n/g' -e 's/[",]//g' | grep "$METRIC_VALUE" | awk -F ':' '{print $2}'
[root@flume112 ~]#
[root@flume112 ~]# ll /etc/zabbix/zabbix_agentd.d/flume_monitor.sh
-rwxr-xr-x root root May : /etc/zabbix/zabbix_agentd.d/flume_monitor.sh
[root@flume112 ~]#
[root@flume112 ~]# chmod +x /etc/zabbix/zabbix_agentd.d/flume_monitor.sh #别忘了添加执行权限
[root@flume112 ~]#
[root@flume112 ~]# ll /etc/zabbix/zabbix_agentd.d/flume_monitor.sh
-rwxr-xr-x root root May : /etc/zabbix/zabbix_agentd.d/flume_monitor.sh
[root@flume112 ~]#
[root@flume112 ~]# /etc/zabbix/zabbix_agentd.d/flume_monitor.sh EventReceivedCount #我们光编写脚本添加了权限还是不够的，我们得手动验证一下脚本是否可用哟！[root@flume112 ~]#
[root@flume112 ~]# /etc/zabbix/zabbix_agentd.d/flume_monitor.sh ChannelFillPercentage
1.7976931348623157E308
[root@flume112 ~]#
[root@flume112 ~]# /etc/zabbix/zabbix_agentd.d/flume_monitor.sh EventAcceptedCount[root@flume112 ~]#
[root@flume112 ~]# /etc/zabbix/zabbix_agentd.d/flume_monitor.sh EventAcceptedCount[root@flume112 ~]#
3>.自定义zabbix agent监控配置文件
[root@flume112 ~]# cat /etc/zabbix/zabbix_agentd.d/flume_monitor.conf
#!/bin/bash
#@author :yinzhengjie
#blog:http://www.cnblogs.com/yinzhengjie
#EMAIL:y1053419035@qq.com
#Data:Thu Oct :: CST UserParameter=FLUME_STATUS[*],/etc/zabbix/zabbix_agentd.d/flume_monitor.sh $1 $
[root@flume112 ~]#
4>.重启zabbix agent服务
[root@flume112 ~]# systemctl restart zabbix-agent
[root@flume112 ~]#
[root@flume112 ~]# systemctl status zabbix-agent
● zabbix-agent.service - Zabbix Agent
Loaded: loaded (/usr/lib/systemd/system/zabbix-agent.service; enabled; vendor preset: disabled)
Active: active (running) since Thu -- :: CST; 6s ago
Process: ExecStop=/bin/kill -SIGTERM $MAINPID (code=exited, status=/SUCCESS)
Process: ExecStart=/usr/sbin/zabbix_agentd -c $CONFFILE (code=exited, status=/SUCCESS)
Main PID: (zabbix_agentd)
CGroup: /system.slice/zabbix-agent.service
├─ /usr/sbin/zabbix_agentd -c /etc/zabbix/zabbix_agentd.conf
├─ /usr/sbin/zabbix_agentd: collector [idle sec]
├─ /usr/sbin/zabbix_agentd: listener # [waiting for connection]
├─ /usr/sbin/zabbix_agentd: listener # [waiting for connection]
├─ /usr/sbin/zabbix_agentd: listener # [waiting for connection]
└─ /usr/sbin/zabbix_agentd: active checks # [idle sec]May :: flume112.aggrx systemd[]: Starting Zabbix Agent...
May :: flume112.aggrx systemd[]: zabbix-agent.service: Supervising process which is not our child. We'll most likely not notice when it exits.
May :: flume112.aggrx systemd[]: Started Zabbix Agent.
[root@flume112 ~]#
三.在zabbix server端进行取值
1>.创建模板
2>.点击目标名称
3>.模板创建完成
4>.为刚刚创建的模板创建监控项（item）
5>.自定义item
6>.添加完成
7>.查看item的内容，clone新的item，重复上面的操作，直到把我们需要监控的item添加完成
8>.模板关键参数监控创建完毕
9>.点击创建主机
10>.添加主机信息
11>.为主机绑定模板，并点击添加按钮
12>.查看该主机的最新数据信息，可以点击旁边的Graph
13>.查看监控图
