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:9092,10.1.2.115:9092,10.1.2.116:9092,10.1.2.117:9092,10.1.2.118:9092 agent.sources.kafkaSource.topic = account-check agent.sources.kafkaSource.kafka.consumer.group.id = 20190507-account-check agent.sources.kafkaSource.kafka.consumer.max.partition.fetch.bytes = 20485760 agent.sources.kafkaSource.kafka.consumer.heartbeat.interval.ms = 120000 agent.sources.kafkaSource.kafka.consumer.rebalance.timeout.ms = 300000 agent.sources.kafkaSource.kafka.consumer.fetch.min.bytes = 10000 agent.sources.kafkaSource.kafka.consumer.session.timeout.ms = 180000 agent.sources.kafkaSource.kafka.consumer.request.timeout.ms = 300000 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:9092,10.1.2.115:9092,10.1.2.116:9092,10.1.2.117:9092,10.1.2.118:9092 agent.channels.kafkaSource.kafka.topic = channel.account-check-20190507-01 agent.channels.kafkaSource.kafka.consumer.group.id = 20190507-channel.account-check-20190507-01 agent.channels.kafkaSource.kafka.consumer.heartbeat.interval.ms = 120000 agent.channels.kafkaSource.kafka.consumer.rebalance.timeout.ms = 300000 agent.channels.kafkaSource.kafka.consumer.fetch.min.bytes = 10000 agent.channels.kafkaSource.kafka.consumer.session.timeout.ms = 180000 agent.channels.kafkaSource.kafka.consumer.request.timeout.ms = 300000 #指定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 = 10-1-2-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 = 0 agent.sinks.hdfsSink.hdfs.rollSize = 0 agent.sinks.hdfsSink.hdfs.rollInterval = 300 agent.sinks.hdfsSink.hdfs.batchSize = 1000 agent.sinks.hdfsSink.hdfs.threadsPoolSize = 25 agent.sinks.hdfsSink.hdfs.idleTimeout = 0 agent.sinks.hdfsSink.hdfs.minBlockReplicas = 1 agent.sinks.hdfsSink.hdfs.callTimeout=100000 agent.sinks.hdfsSink.hdfs.request-timeout=100000 agent.sinks.hdfsSink.hdfs.connect-timeout=80000 [root@flume112 ~]#
[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 18 11:26:06 CST 2018 #启动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=11201 -Dflume.root.logger=INFO,console >> /soft/flume/logs/flume-http-p2p01.log 2>&1 & [root@flume112 ~]#
2>.我们启动上面的Flume agent后
[root@flume112 ~]# curl 10.1.2.112:11201/metrics | jq % Total % Received % Xferd Average Speed Time Time Time Current Dload Upload Total Spent Left Speed 1090 0 1090 0 0 165k 0 --:--:-- --:--:-- --:--:-- 177k { "SINK.hdfsSink": { #这是HDFS SINK监控数据。 "ConnectionCreatedCount": "77", #下一个阶段（或存储系统）创建链接的数量（如HDFS创建一个文件）。 "ConnectionClosedCount": "76", #下一个阶段（或存储系统）关闭链接的数量（如HDFS中关闭一个文件）。 "Type": "SINK", #很显然，这里是SINK监控项，类型为SINK "BatchCompleteCount": "24891", #批量处理event的个数等于批处理大小的数量。 "BatchEmptyCount": "3624", #批量处理event的个数为0的数量（空的批量的数量），如果数量很大表示source写入数据的速度比sink处理数据的速度慢很多。 "EventDrainAttemptCount": "24960633", #sink尝试写出到存储的事件总数量。 "StartTime": "1557363309342", #channel启动时的毫秒值时间。 "EventDrainSuccessCount": "24960633", #sink成功写出到存储的事件总数量。 "BatchUnderflowCount": "155", #批量处理event的个数小于批处理大小的数量（比sink配置使用的最大批量尺寸更小的批量的数量），如果该值很高也表示sink比source更快。 "StopTime": "0", #channel停止时的毫秒值时间，为0表示一直在运行。 "ConnectionFailedCount": "0" #下一个阶段（或存储系统）由于错误关闭的连接数量（如HDFS上一个新创建的文件由于超市而关闭）。 }, "CHANNEL.kafkaSource": { #这是KAFKA SOURCE监控数据。 "KafkaEventGetTimer": "2712566", #KAFKA事件获取计数器。 "ChannelSize": "0", #目前channel中事件的总数量，目前仅支持File Channel，Memory channel的统计数据。我这里使用的是Kafka channel。 "EventTakeAttemptCount": "0", #sink尝试从channel拉取事件的总次数。这不意味着每次时间都被返回，因为sink拉取的时候channel可能没有任何数据。 "StartTime": "1557363309336", #channel启动时的毫秒值时间。 "KafkaCommitTimer": "21362", #KAFKA提交计数器。 "ChannelCapacity": "0", #channel的容量，目前仅支持File Channel，Memory channel的统计数据。我这里使用的是Kafka channel。 "ChannelFillPercentage": "1.7976931348623157E308", #channel已填入的百分比。 "Type": "CHANNEL", #很显然，这里是CHANNEL监控项，类型为CHANNEL。 "EventTakeSuccessCount": "24960633", #sink成功从channel读取事件的总数量。 "RollbackCount": "0", #回滚的次数。 "KafkaEventSendTimer": "1075974", #KAFKA事件发送计数器。 "EventPutAttemptCount": "0", #Source尝试写入Channe的事件总次数。 "EventPutSuccessCount": "24797364", #成功写入channel且提交的事件总次数。 "StopTime": "0" #channel停止时的毫秒值时间，为0表示一直在运行。 }, "SOURCE.kafkaSource": { "KafkaEventGetTimer": "3274677", #KAFKA事件获取计数器。 "AppendBatchAcceptedCount": "0", #成功提交到channel的批次的总数量。 "AppendReceivedCount": "0", #每批只有一个事件的事件总数量（与RPC调用的一个append调用相等）。 "EventAcceptedCount": "24797364", #成功写出到channel的事件总数量。 "StartTime": "1557363427104", #source启动时的毫秒值时间。 "AppendBatchReceivedCount": "0", #接收到事件批次的总数量。 "KafkaCommitTimer": "75112", #KAFKA提交计数器。 "EventReceivedCount": "24797364", #目前为止source已经接收到的事件总数量。 "Type": "SOURCE", #很显然，这里是SOURCE监控项，类型为SOURCE。 "KafkaEmptyCount": "0", #KAFKA空的批量的数量。 "AppendAcceptedCount": "0", #逐条录入的次数，单独传入的事件到Channel且成功返回的事件总数量。 "OpenConnectionCount": "0", #目前与客户端或sink保持连接的总数量，目前仅支持avro source展现该度量。 "StopTime": "0" #source停止时的毫秒值时间，为0表示一直在运行。 } } [root@flume112 ~]#
温馨提示：
上图需要注意的点我已经用深颜色标识出来啦，当然，如果你还要想了解更多度量值，可参考官方文档：http://flume.apache.org/FlumeUserGuide.html#monitoring。
3>.上面是使用jq工具查看Flume自身监控的，当然，我们也可以使用sed命令将上面的内容格式化一下
[root@flume112 ~]# curl 10.1.2.112:11201/metrics 2>/dev/null |sed -e 's/([,])s*/1 /g' -e 's/[{}]/ /g' -e 's/[",]//g' SINK.hdfsSink: ConnectionCreatedCount:91 ConnectionClosedCount:90 Type:SINK BatchCompleteCount:28856 BatchEmptyCount:4291 EventDrainAttemptCount:28938850 StartTime:1557363309342 EventDrainSuccessCount:28938850 BatchUnderflowCount:190 StopTime:0 ConnectionFailedCount:0 CHANNEL.kafkaSource: KafkaEventGetTimer:3153432 ChannelSize:0 EventTakeAttemptCount:0 StartTime:1557363309336 KafkaCommitTimer:24248 ChannelCapacity:0 ChannelFillPercentage:1.7976931348623157E308 Type:CHANNEL EventTakeSuccessCount:28938850 RollbackCount:0 KafkaEventSendTimer:1258095 EventPutAttemptCount:0 EventPutSuccessCount:28775581 StopTime:0 SOURCE.kafkaSource: KafkaEventGetTimer:3778649 AppendBatchAcceptedCount:0 AppendReceivedCount:0 EventAcceptedCount:28775581 StartTime:1557363427104 AppendBatchReceivedCount:0 KafkaCommitTimer:85960 EventReceivedCount:28775581 Type:SOURCE KafkaEmptyCount:0 AppendAcceptedCount:0 OpenConnectionCount:0 StopTime:0 [root@flume112 ~]# [root@flume112 ~]# curl 10.1.2.112:11201/metrics 2>/dev/null |sed -e 's/([,])s*/1 /g' -e 's/[{}]/ /g' -e 's/[",]//g' | grep EventDrainSuccessCount #我们可以进一步拿到某个监控参数！ EventDrainSuccessCount:28938850 [root@flume112 ~]#
4>.使用awk获取某个参数的值
[root@flume112 ~]# curl 10.1.2.112:11201/metrics 2>/dev/null |sed -e 's/([,])s*/1 /g' -e 's/[{}]/ /g' -e 's/[",]//g' | grep EventDrainSuccessCount EventDrainSuccessCount:29215653 [root@flume112 ~]# [root@flume112 ~]# [root@flume112 ~]# curl 10.1.2.112:11201/metrics 2>/dev/null |sed -e 's/([,])s*/1 /g' -e 's/[{}]/ /g' -e 's/[",]//g' | grep EventDrainSuccessCount | awk -F ':' '{print $2}' 29215653 [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 18 11:26:06 CST 2018 FLUME_PORT=$1 METRIC_VALUE=$2 curl 127.0.0.1:"$FLUME_PORT"/metrics 2>/dev/null |sed -e 's/([,])s*/1 /g' -e 's/[{}]/ /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 1 root root 305 May 9 16:55 /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 1 root root 305 May 9 16:55 /etc/zabbix/zabbix_agentd.d/flume_monitor.sh [root@flume112 ~]# [root@flume112 ~]# /etc/zabbix/zabbix_agentd.d/flume_monitor.sh 11201 EventReceivedCount #我们光编写脚本添加了权限还是不够的，我们得手动验证一下脚本是否可用哟！ 32172360 [root@flume112 ~]# [root@flume112 ~]# /etc/zabbix/zabbix_agentd.d/flume_monitor.sh 11201 ChannelFillPercentage 1.7976931348623157E308 [root@flume112 ~]# [root@flume112 ~]# /etc/zabbix/zabbix_agentd.d/flume_monitor.sh 11201 EventAcceptedCount 32172360 [root@flume112 ~]# [root@flume112 ~]# /etc/zabbix/zabbix_agentd.d/flume_monitor.sh 11202 EventAcceptedCount 58744826 [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 18 11:26:06 CST 2018 UserParameter=FLUME_STATUS[*],/etc/zabbix/zabbix_agentd.d/flume_monitor.sh $1 $2 [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 2019-05-09 16:59:31 CST; 6s ago Process: 14476 ExecStop=/bin/kill -SIGTERM $MAINPID (code=exited, status=0/SUCCESS) Process: 14479 ExecStart=/usr/sbin/zabbix_agentd -c $CONFFILE (code=exited, status=0/SUCCESS) Main PID: 14481 (zabbix_agentd) CGroup: /system.slice/zabbix-agent.service ├─14481 /usr/sbin/zabbix_agentd -c /etc/zabbix/zabbix_agentd.conf ├─14482 /usr/sbin/zabbix_agentd: collector [idle 1 sec] ├─14483 /usr/sbin/zabbix_agentd: listener #1 [waiting for connection] ├─14484 /usr/sbin/zabbix_agentd: listener #2 [waiting for connection] ├─14485 /usr/sbin/zabbix_agentd: listener #3 [waiting for connection] └─14486 /usr/sbin/zabbix_agentd: active checks #1 [idle 1 sec] May 09 16:59:31 flume112.aggrx systemd[1]: Starting Zabbix Agent... May 09 16:59:31 flume112.aggrx systemd[1]: zabbix-agent.service: Supervising process 14481 which is not our child. We'll most likely not notice when it exits. May 09 16:59:31 flume112.aggrx systemd[1]: 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>.查看监控图
