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

时间:2021-06-01 10:21:17

需求背景: 

  在使用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