flume自定义拦截器实现添加IP

需求背景： 

　　在使用flume收集日志的时候，需要将本机的IP地址添加到body中再包装在event中，以便后期的日志分类，和快速的故障定位，但是没有发现有现成的interceptor来实现以上功能。

实现步骤：

　　1：新建一个类，实现Interceptor接口

　　2：重写intercept(Event arg0)方法

　　3：打成jar包，放入Flume_Home 目录的lib文件夹下

　　4： 修改flume配置文件，将拦截器的type值指向自定义类的地址，如果有自定义属性，需要配置该自定义属性

源码如下：

　　1： 自定义类      AppendIPInterceptor

package com.xxx.flume.interceptor;

/**
 * Created by vincent.yu on 2018/07/23.
 */
import com.google.common.base.Charsets;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.util.List;

public class AppendIPInterceptor implements Interceptor {

    private String serviceId=null;
// 自定义属性 serviceId
    public AppendIPInterceptor(String _serviceId){
        serviceId=_serviceId;
    }

    public Event intercept(Event arg0) {
        String eventBody = new String(arg0.getBody(),Charsets.UTF_8);
        String fmt="%s %s";
// 添加serviceId 到event的开头
        arg0.setBody(String.format(fmt, serviceId,eventBody).getBytes());
        return arg0;
    }

    public List<Event> intercept(List<Event> events) {
        for (Event event : events) {
            intercept(event);
        }
        return events;
    }

    public void close() {
        //~ null
    }

    public void initialize() {
        //~ null
    }

    /*public static class Builder implements Interceptor.Builder {

        public void configure(Context context) {

        }

        public Interceptor build() {
            return new AppendIPInterceptor();
        }
    } */
}

　　2： 自定义 Builder ---> AppendIPBuilder

package com.xxx.flume.interceptor;

import org.apache.flume.Context;
import org.apache.flume.interceptor.Interceptor;

/**
 * Created by vincent.yu on 2018/07/23.
 */
public class AppendIPBuilder implements Interceptor.Builder{

    private String serviceId=null;
    public void configure(Context context) {
//       set argument serviceId
        String configServiceId=context.getString("serviceId");
        serviceId=configServiceId;
    }

    public Interceptor build() {
        return new AppendIPInterceptor(serviceId);
    }

}

　　3： 修改flume配置文件

a1.sources.r1.interceptors =  i2
#a1.sources.r1.interceptors.i1.type = com.xxx.flume.interceptor.MyMultiLineExecBuilder
a1.sources.r1.interceptors.i2.type = com.xxx.flume.interceptor.AppendIPBuilder
a1.sources.r1.interceptors.i2.serviceId = [source_ip]10.0.53.237