正文
Flume-NG内置计数器(监控)源码级分析
小程序：扫一扫查出行【扫一扫了解最新限行尾号】
复制小程序
Flume的内置监控怎么整？这个问题有很多人问。目前了解到的信息是可以使用Cloudera Manager、Ganglia有图形的监控工具，以及从浏览器获取json串，或者自定义向其他监控系统汇报信息。那监控的信息是什么呢？就是各个组件的统计信息，比如成功接收的Event数量、成功发送的Event数量，处理的Transaction的数量等等。而且不同的组件有不同的Countor来做统计，目前直到1.5版本仍然只对三大组件：source、sink、channel进行统计分别是SourceCounter、SinkCounter、ChannelCounter，这三个计数器的统计项是固定的，就是你不能自己设置自己的统计项；另外还有ChannelProcessorCounter和SinkProcessorCounter，这两项目前没有设置统计项，所以是目前还是“摆设”。另外有些同学可能也发现了，有些内置的组件使用CounterGroup这个来统计信息，这个可以自己随意设置统计项，但是遗憾的是目前(1.5版本)这个可以自定义的计数器的信息还无法用在监控上，因为这只是一个单独的类，并没有继承MonitoredCounterGroup这个抽象类。有些内置组件使用的是CounterGroup，所以监控时会没有数据，不同的版本使用此CounterGroup的组件可能不同。下面我们重点介绍：SourceCounter、SinkCounter、ChannelCounter。
Flume-NG的所有统计信息、监控及相关的类都在org.apache.flume.instrumentation.http、org.apache.flume.instrumentation、org.apache.flume.instrumentation.util三个包下。
上面提到了MonitoredCounterGroup，这个类是用来跟踪内部的统计指标的，注册组件的MBean并跟踪和更新统计值。需要监控的组件都要继承这个类，这个类可以跟踪flume内部的所有组件，但是目前只实现了3个。其中比较重要的方法有以下几个：
(1)、构造方法MonitoredCounterGroup(Type type, String name, String... attrs)，这个方法主要是设置组件的类型、名称；然后将所有的attrs(这是设定的各个统计项)加入Map<String, AtomicLong> counterMap，值设定为0；然后初始化计数器的开始时间和结束时间，都设为0.
(2)、start()方法，会先注册计数器，然后对所有统计项的统计值设为0；将开始时间设置为当前时间
(3)、register()方法，如果这个计数器还未注册，将这个计数器的MBean进行注册，就可以进行跟踪了
(4)、stop()方法，会设置结束时间为当前时间；输出各个统计项的信息。我们 Ctrl+C 结束进程时，最后显示的统计信息就是来自这里。
其它方法都是获取counterMap的中信息或者更新值等，比较简单。
接下来我们看看，三个组件中各种统计项及其含义吧：
一、SourceCounter，继承了MonitoredCounterGroup。主要统计项如下：
(1)"src.events.received"，表示source接受的event个数；
(2)"src.events.accepted"，表示source处理成功的event个数，和上面的区别就是上面虽然接受了可能没处理成功；
(3)"src.append.received"，表示调用append次数，在avrosource和thriftsource中调用；
(4)"src.append.accepted"，表示append处理成功次数；
(5)"src.append-batch.received"，表示appendBatch被调用的次数，在avrosource和thriftsource中调用；
(6)"src.append-batch.accepted"，表示appendBatch处理成功次数；
(7)"src.open-connection.count"，用在avrosource中表示打开连接的数量；
一般source调用都集中在前俩。
二、SinkCounter，继承了MonitoredCounterGroup
(1)"sink.connection.creation.count"，这个调用的地方颇多，都表示“链接”创建的数量，比如与HBase建立链接，与avrosource建立链接以及文件的打开等；
(2)"sink.connection.closed.count"，对应于上面的stop操作、destroyConnection、close文件操作等。
(3)"sink.connection.failed.count"，表示上面所表示“链接”时异常、失败的次数；
(4)"sink.batch.empty"，表示这个批次处理的event数量为0的情况；
(5)"sink.batch.underflow"，表示这个批次处理的event的数量介于0和配置的batchSize之间；
(6)"sink.batch.complete"，表示这个批次处理的event数量等于设定的batchSize；
(7)"sink.event.drain.attempt"，准备处理的event的个数；
(8)"sink.event.drain.sucess"，这个表示处理成功的event数量，与上面不同的是上面的是还未处理的。
三、ChannelCounter，继承了MonitoredCounterGroup
(1)"channel.current.size"，这个表示这个channel的当前容量；
(2)"channel.event.put.attempt"，一般指的是在channel的事务当中，source的put操作中记录尝试发送event的个数；
(3)"channel.event.take.attempt"，一般指的是在channel的事务中，sink的take操作记录尝试拿event的个数；
(4)"channel.event.put.success"，一般指的是在channel的事务中，put成功的event的数量；
(5)"channel.event.take.success"，一般指的是channel事务中，take成功的event的数量；
(6)"channel.capacity"，指的是channel的容量，在channel的start方法中设置。
上面这些统计项都是固定的，我们可以根据需要增加相应项的值，可以在监控中查看组件的变化情况，从而掌握flume进程的运行情况。比如可以查看channel的容量从而了解到source和sink的相对处理速度，还有可以看source或者sink每个批次处理成功与失败的次数，了解组件的运行状况等等。
当然有些同学可能在自定义自己的组件时，想统计一些自己的统计项，这些统计项在上面三大组件中是没有，怎么办？自己定制啊，上面说了必须要继承MonitoredCounterGroup这个抽象类，设定自己的统计项，然后将统计项设置成数组调用MonitoredCounterGroup的构造函数；然后在自定义的计数器中增加更新数值的方法。最后在自定义的组件中构造自定义的计数器，并启用它的start方法，剩下的就是在该更新统计项数值的地方更新就可以了。
还有一个重要的内容就是监控的实现！没错，内置的有两种HTTP方式(就是json串)和Ganglia，后者需要安装Ganglia，前者非常简单，只需要在Flume的启动命令中加上：-Dflume.monitoring.type=http -Dflume.monitoring.port=XXXX ，最后的XXXX是你需要设置的端口！然后你就可以在浏览器上通过访问这个Flume所在节点的IP：XXXX/metrics，不断刷新就可以看到最新的组件统计信息。关于Ganglia的请读者自行组建Ganglia集群并参考用户指南来操作。
如果我想自己实现一个server向其他系统汇报信息，咋整？目前有至少两个方法：
一、就是上面的HTTP啊，你可以不断去获取json串，自己解析出来各个统计指标，然后剩下的就是你想怎么整就怎么整吧。
二、就是自己实现一个类似HTTP的server，必须实现org.apache.flume.instrumentation.MonitorService接口，这个接口只有俩方法:start和stop。这个接口继承自Configurable接口所以拥有可以读取配置文件的configure(configure(Context context))方法，来获取一些配置信息。
以HTTP为例(对应的类是org.apache.flume.instrumentation.http.HTTPMetricsServer)，它的start方法启动了一个jetty作为web server，提供WEB服务。并实现了AbstractHandler的一个处理数据的类HTTPMetricsHandler，这个类的handle(String target, HttpServletRequest request, HttpServletResponse response,int dispatch)方法来设置一些WEB页面的格式以及通过JMXPollUtil.getAllMBeans()获取所有组件注册的MBean构成的Map<String, Map<String, String>> metricsMap，遍历这个metricsMap将这个metricsMap转换成json输出到web页面。stop方法就是一些清理工作，这里是关闭jetty server。很简单吧，所以我们完全可以实现一个server，在start方法中启动一个线程每隔一秒或者自己定遍历这个metricsMap，写入mysql、HBase或者别的地方，你随便。。。
你可以在定义的组件中调用自己的计数器，然后将计数器、监控类、自定义组件(source、sink、channel)打包放到lib下，在启动命令后加-Dflume.monitoring.type=AAAAA -Dflume.monitoring.node=BBBB，就可以了。注意，Dflume.monitoring.type这个好似必须要设置的，就是你自己的监控类(这里是AAAAA)，后面的可有可无都是一些参数，你可以自定义参数名，比如可以设置数据库服务器IP、端口等。
至此，这里介绍完了。这些都是从源码中看出来的，还未曾实现，供大伙借鉴。
