对于flume的监控,只需要监控channel的性能数据即可,source和sink的性能一部分可以从channel中表现出来。
以MemoryChannel为例,在MemoryTransaction的构造函数中会实例化一个org.apache.flume.instrumentation.ChannelCounter对象
public MemoryTransaction( int transCapacity, ChannelCounter counter) { putList = new LinkedBlockingDeque<Event>(transCapacity); takeList = new LinkedBlockingDeque<Event>(transCapacity); channelCounter = counter; }
org.apache.flume.instrumentation.ChannelCounter 定义了几个计数器用来记录channel的性能数据
private static final String COUNTER_CHANNEL_SIZE = "channel.current.size"; //已经使用的容量大小 private static final String COUNTER_EVENT_PUT_ATTEMPT = "channel.event.put.attempt"; //source到channel尝试插入的数据(不管是否成功) private static final String COUNTER_EVENT_TAKE_ATTEMPT = "channel.event.take.attempt"; //sink从channel尝试消费的数据(不管是否成功) private static final String COUNTER_EVENT_PUT_SUCCESS = "channel.event.put.success"; //source到channel成功插入的数据 private static final String COUNTER_EVENT_TAKE_SUCCESS = "channel.event.take.success"; //sink从channel成功消费的数据 private static final String COUNTER_CHANNEL_CAPACITY = "channel.capacity"; //总容量大小
并封装了相关的方法,来操作这些性能计数器:
比如channel.event.put.attempt的由getEventPutAttemptCount和incrementEventPutAttemptCount操作:
public long incrementEventPutAttemptCount() { //用于数量增加1 return increment( COUNTER_EVENT_PUT_ATTEMPT); } public long getEventPutAttemptCount() { //用于获取值 return get( COUNTER_EVENT_PUT_ATTEMPT); }
而在channel的相关操作中会使用到这些方法:
protected void doPut(Event event) throws InterruptedException { channelCounter.incrementEventPutAttemptCount(); //比如在插入数据的操作开始时,增加channel.event.put.attempt的值 int eventByteSize = (int)Math.ceil(estimateEventSize(event)/ byteCapacitySlotSize); if (! putList.offer(event)) { throw new ChannelException( "Put queue for MemoryTransaction of capacity " + putList.size() + " full, consider committing more frequently, " + "increasing capacity or increasing thread count" ); } putByteCounter += eventByteSize; }
counter的注册使用,以memorychannel相关为例:
ChannelCounter扩展了MonitoredCounterGroup类并实现了ChannelCounterMBean接口
MonitoredCounterGroup是一个抽象类,其具体的实现类定义了具体的组件的性能计数器和对应的封装方法
ChannelCounter中包含的所有的可用的counter:
private static final String[] ATTRIBUTES = { COUNTER_CHANNEL_SIZE, COUNTER_EVENT_PUT_ATTEMPT, COUNTER_EVENT_TAKE_ATTEMPT, COUNTER_EVENT_PUT_SUCCESS, COUNTER_EVENT_TAKE_SUCCESS, COUNTER_CHANNEL_CAPACITY };
ChannelCounter的构造方法调用MonitoredCounterGroup的构造方法:
public ChannelCounter(String name) { super(MonitoredCounterGroup.Type. CHANNEL, name, ATTRIBUTES ); //调用MonitoredCounterGroup构造方法 } MonitoredCounterGroup构造方法: private final Map<String, AtomicLong> counterMap; .... protected MonitoredCounterGroup(Type type, String name, String... attrs) { this. type = type; this. name = name; Map<String, AtomicLong> counterInitMap = new HashMap<String, AtomicLong>(); // 声明一个初始的hashmap,用来存放counter name到value的对应关系 // Initialize the counters for (String attribute : attrs) { counterInitMap.put(attribute, new AtomicLong(0L)); //初始value都为0 } counterMap = Collections.unmodifiableMap(counterInitMap); //返回hashmap不可更改的映射视图 startTime = new AtomicLong(0L); stopTime = new AtomicLong(0L); }
这里Type是一个enum类型,可取值:
public static enum Type { SOURCE, CHANNEL_PROCESSOR, CHANNEL, SINK_PROCESSOR, SINK, INTERCEPTOR, SERIALIZER, OTHER };
在MemoryChannel中的start方法中启动:
public synchronized void start() { channelCounter.start(); //调用MonitoredCounterGroup的start方法 channelCounter.setChannelSize( queue.size()); channelCounter.setChannelCapacity(Long. valueOf( queue.size() + queue.remainingCapacity())); super.start(); }
MonitoredCounterGroup.start:
public void start() { register(); //调用register方法注册counter,主要是调用ManagementFactory. getPlatformMBeanServer().registerMBean( this, objName);进行注册mbean操作,把ChannelCounter对象作为mbean进行注册 stopTime.set(0L); for (String counter : counterMap.keySet()) { counterMap.get(counter).set(0L); //设置值都为0 } startTime.set(System. currentTimeMillis()); logger.info( "Component type: " + type + ", name: " + name + " started" ); }
这样就可以通过jmx获取的监控项
时间: 2024-10-08 10:04:20