flume自定义拦截器实现定制收集日志需求

不念不忘少年蓝@ 2024-04-17 17:39 59阅读 0赞

flume默认提供了timestamp,host,static,regex等几种类型的拦截器，timestamp,host,static等拦截器，其实就是在消息头中增加了时间戳，主机名，键值对信息，这些信息可以作用于信宿中。比如有时间戳的话，我们可以存储消息的时候，按照日期文件夹的形式来存放每天的日志。这些拦截器功能有时候，不一定能够满足所有用户需求，因此flume支持用户自定义拦截器，来更加精确的收集日志信息。

flume自定义拦截器，需要继承Interceptor接口，并实现相关方法，同时还需要自定义Builder，返回我们的自定义拦截器。下面看具体的代码，因为自定义flume拦截器，需要加入flume依赖，我们加入flume-ng-core(org.apache.flume)依赖，如果是maven工程，我们可以在pom.xml配置文件中加入如下配置：

<dependency>
      <groupId>org.apache.flume</groupId>
      <artifactId>flume-ng-core</artifactId>
      <version>1.9.0</version>
    </dependency>

我这里使用的是flume版本是1.9.0，所以对应加入的依赖也是1.9.0版本。自定义拦截器代码如下：

package com.xxx.flume;
    import java.io.ByteArrayOutputStream;
    import java.util.ArrayList;
    import java.util.List;
    import java.util.Map;
    import org.apache.flume.Context;
    import org.apache.flume.Event;
    import org.apache.flume.interceptor.Interceptor;
    /***
     * 拦截器：判断消息体是否以"aa"开头，如果是，返回为新的消息体，如果不是，返回空。
     * 新的消息体：aaxxx	timestamp=15xxxxxxxxxxx	host=localhost	type=type
     */
    public class ZoneInterceptor implements Interceptor{
    
    	@Override
    	public void close() {
    		// TODO Auto-generated method stub
    		
    	}
    
    	@Override
    	public void initialize() {
    		// TODO Auto-generated method stub
    		
    	}
    
    	@Override
    	public Event intercept(Event event) {
    		// TODO Auto-generated method stub
    		try {
    			Map<String, String> header = event.getHeaders();
    			String str = new String(event.getBody(),"UTF-8");
    			String timestamp = "";
    			String host = "";
    			String type="";
    			if(header.containsKey("timestamp")){
    				timestamp = "timestamp="+header.get("timestamp");
    			}
    			if(header.containsKey("host")){
    				host = "host="+header.get("host");
    			}
    			
    			if(header.containsKey("type")){
    				type="type="+header.get("type");
    			}
    			if(str.startsWith("aa")){
    				ByteArrayOutputStream output = new ByteArrayOutputStream();
    				output.write(event.getBody());
    				output.write("\t".getBytes());
    				output.write(timestamp.getBytes());
    				output.write("\t".getBytes());
    				output.write(host.getBytes());
    				output.write("\t".getBytes());
    				output.write(type.getBytes());
    				event.setBody(output.toByteArray());
    				return event;
    			}
    		} catch (Exception e) {
    			// TODO: handle exception
    		}
    		return null;
    	}
    
    	@Override
    	public List<Event> intercept(List<Event> list) {
    		// TODO Auto-generated method stub
    		List<Event> result = new ArrayList<Event>();
    		for(Event event:list){
    			Event e = intercept(event);
    			if(e!=null){
    				result.add(e);
    			}
    		}
    		return result;
    	}
    
    	public static class Builder implements Interceptor.Builder{
    
    		@Override
    		public void configure(Context context) {
    			// TODO Auto-generated method stub
    			
    		}
    
    		@Override
    		public Interceptor build() {
    			// TODO Auto-generated method stub
    			return new ZoneInterceptor();
    		}
    		
    	}
    }

简单介绍一下这个拦截器的功能，我们假定收集来自网络消息，判断消息是否以aa开头，如果是，则将消息收集，否则放弃。收集的消息，我们再进行一次改变，在消息体后面通过制表符隔开的方式，新增时间戳，主机，type=flume内容。时间戳，主机和type这些信息来源于其他的拦截器，timestamp,host,static这三个拦截器。我们从消息头部的映射中得到这些变量的值，然后添加到消息体中。

自定义拦截器的思路，就是实现intercept(Event event)方法和intercept(List<Event> list)方法。我们可以在intercept(Event event) 方法中做判断，然后拼接消息体。

最重要的是，我们还需要定义一个静态Builder类实现Interceptor.Builder接口，然后在它的方法build()中返回我们自定义的拦截器。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaW5pZmk_size_16_color_FFFFFF_t_70][]

打包自定义拦截器代码，然后上传到flume安装目录下的lib目录，然后我们来做一个配置文件：

aa.sources=netsource
    aa.channels=c
    aa.sinks=k
    
    aa.sources.netsource.type=netcat
    aa.sources.netsource.bind=0.0.0.0
    aa.sources.netsource.port=30000
    aa.sources.netsource.interceptors=timestamp host static zi
    aa.sources.netsource.interceptors.timestamp.type=timestamp
    aa.sources.netsource.interceptors.host.type=host
    aa.sources.netsource.interceptors.static.type=static
    aa.sources.netsource.interceptors.static.key=type
    aa.sources.netsource.interceptors.static.value=flume
    aa.sources.netsource.interceptors.zi.type=com.xxx.flume.ZoneInterceptor$Builder
    
    aa.channels.c.type=memory
    aa.channels.c.capacity=1000
    aa.channels.c.transactionCapacity=100
    aa.sinks.k.type=FILE_ROLL
    aa.sinks.k.sink.directory=/home/software/flume/files
    aa.sinks.k.sink.rollInterval=0
    aa.sinks.k.sink.rollCount=3
    
    aa.sources.netsource.channels=c
    aa.sinks.k.channel=c

在自定义拦截器前面，我们增加了三个拦截器分别是timestamp,host,static，这些拦截器会将timestamp,host,type三个变量加入消息头的map中，我们在自定义拦截器中就正好可以使用这些变量。

启动flume,加载配置文件：

bin/flume-ng agent -c conf -f conf/custom\_interceptors.conf -n aa -Dflume.root.logger=INFO,console

启动成功，通过telnet连接127.0.0.1 30000，发送5条消息，其中3条是满足拦截器的条件的，即以"aa"开头的消息：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaW5pZmk_size_16_color_FFFFFF_t_70 1][]

发送成功，我们通过查看保存的文件内容，看看拦截器的效果，保存的文件中，只有三条以aa开头的消息，并且后面拼接了timestamp,host,type信息：

![20190829145515485.png][]

至此，一个简单的flume自定义拦截器就实现了，并且通过测试，验证了拦截器的正确性。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaW5pZmk_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/9a780437e36842ddb6388d3605d07440.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaW5pZmk_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/fcda8d2d634941afa946cf8a12cada5a.png
[20190829145515485.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/7ecef41cc46c424e9abe68fef601ac22.png