Flume NetCat UDP Source 获取客户端IP

待我称王封你为后i 2022-12-31 13:22 48阅读 0赞

> Flume-ng version: 1.9.0
> 
> Flume 官方文档：[http://flume.apache.org/][http_flume.apache.org]

\- 官方文档中关于 NetCat UDP Source 的说明如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDIzODYyNQ_size_16_color_FFFFFF_t_70][]

其中 **remoteAddressHeader** 这一项就是指定一个 key 用于在header中存放客户端IP，但实际使用中发现存在出入。

\- 先编写一个自定义拦截器，将接收的event header和body在debug模式下打印出来看下，拦截器代码如下：

package com.mit.DC;
    
    import java.nio.charset.StandardCharsets;
    import java.util.List;
    
    import java.text.ParseException;
    import java.util.*;
    
    import org.apache.flume.Context;
    import org.apache.flume.Event;
    import org.apache.flume.interceptor.Interceptor;
    
    import com.alibaba.fastjson.JSON;
    
    
    /**
     * @Author mit
     * @Description syslog 数据自定义拦截器
     * @Date 2020/12/29 下午7:28
     * @Version 1.0
     */
    public class SyslogDataInterceptor implements Interceptor {
    
        @Override
        public void initialize(){
    
        }
    
        /*
         * single event interceptor
         */
        @Override
        public Event intercept(Event event) {
            byte[] body = event.getBody();
            Map<String, String> headers = event.getHeaders();
            byte[] newBody = new byte[0];
            String bodyStr = new String(body, StandardCharsets.UTF_8);
            System.out.println("111111111111111111111111111111111111111111111111"); // 方便定位输出的数据；
            System.out.println(JSON.toJSONString(headers));
            System.out.println(bodyStr);
            System.out.println("222222222222222222222222222222222222222222222222");
            event.setBody(newBody);
            return event;
        }
    
        /*
         * batch event interceptor
         */
        @Override
        public List<Event> intercept(List<Event> eventList) {
            for (Event event : eventList) {
                intercept(event);
            }
            return eventList;
        }
    
        @Override
        public void close() {
    
        }
    
        /*
         * build interceptor
         */
        public static class Builder implements Interceptor.Builder {
            /*
             * get config from config file
             */
            @Override
            public void configure(Context context) {
    
            }
    
            /*
             * create interceptor object
             * */
            @Override
            public Interceptor build() {
                return new SyslogDataInterceptor();
            }
        }
    
    }

\- flume-ng 配置文件 udp\_kafka.conf ，我这里采用的是kafka channel 无sinker的拓扑，数据接收后就直接入kafka了，后面的计算节点再消费kafka数据。

\- 将客户端IP在header中的key指定为 **clientIP**

a1.sources = r1
    a1.channels = c1
    
    a1.sources.r1.type=netcatudp
    a1.sources.r1.host=0.0.0.0
    a1.sources.r1.port=6001
    a1.sources.r1.remoteAddressHeader=clientIP
    a1.sources.r1.channels=c1
    
    a1.sources.r1.interceptors = i1
    a1.sources.r1.interceptors.i1.type = com.mit.DC.SyslogDataInterceptor$Builder
    
    a1.channels.c1.type = org.apache.flume.channel.kafka.KafkaChannel
    a1.channels.c1.kafka.bootstrap.servers = kafka-01:9092,kafka-02:9092,kafka-03:9092
    a1.channels.c1.kafka.topic = RawData
    a1.channels.c1.parseAsFlumeEvent = false

\- 接下来使用 flume-ng 的debug模式看下 event header 和 body 数据

\- flume-ng debug 模式

flume-ng agent -c /usr/local/flume/conf/ -f /usr/local/flume/conf/udp_kafka.conf -n a1 -Dflume.root.logger=DEBUG,console

\- 使用nc工具向flume 监听的6001端口发送数据

![20201229234557504.png][]

\- 查看flume-ng打印出来的数据

![2020122923470077.png][]

可以看到这里的key是 **null** 并且客户端IP前还多了个 **/**

\- 查看下flume-ng的源码，可以看到 NetCatUdpSource.java中关于客户端IP的配置项

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDIzODYyNQ_size_16_color_FFFFFF_t_70 1][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDIzODYyNQ_size_16_color_FFFFFF_t_70 2][]

现在就很清楚了，是**flume-ng源码与文档间存在出入，我们按源码中定义的 "****remoteAddress****" 来配置客户端IP在header中使用的key。**

\- 修改udp\_kafka.conf 配置如下

a1.sources = r1
    a1.channels = c1
    
    a1.sources.r1.type=netcatudp
    a1.sources.r1.host=0.0.0.0
    a1.sources.r1.port=6001
    a1.sources.r1.remoteAddress=clientIP
    a1.sources.r1.channels=c1
    
    a1.sources.r1.interceptors = i1
    a1.sources.r1.interceptors.i1.type = com.mit.DC.SyslogDataInterceptor$Builder
    
    a1.channels.c1.type = org.apache.flume.channel.kafka.KafkaChannel
    a1.channels.c1.kafka.bootstrap.servers = kafka-01:9092,kafka-02:9092,kafka-03:9092
    a1.channels.c1.kafka.topic = RawData
    a1.channels.c1.parseAsFlumeEvent = false

\-  再次debug运行看下效果

![20201229235618768.png][]

![20201229235653443.png][]

可以看到修改配置项为 remoteAddress 后成功在header中将客户端IP放在了指定的key下

\- 自定义拦截器将客户端IP添加到body中，其它代码与上面的一致，修改重写的 intercept 如下：

/*
     * single event interceptor
     */
    @Override
    public Event intercept(Event event) {
        byte[] body = event.getBody();
        Map<String, String> headers = event.getHeaders();
        byte[] newBody = new byte[0];
        String bodyStr = new String(body, StandardCharsets.UTF_8);
        // 將header中的客戶端IP拼接到body中
        String newBodyStr = String.join(", ", headers.get("clientIP").replace("/", ""), bodyStr);
        System.out.println("111111111111111111111111111111111111111111111111");
        System.out.println(newBodyStr);
        System.out.println("222222222222222222222222222222222222222222222222");
        newBody = newBodyStr.getBytes(StandardCharsets.UTF_8);
        event.setBody(newBody);
        return event;
    }

\- 运行查看下效果

![20201230000230704.png][]

![20201230000252613.png][]

[http_flume.apache.org]: http://flume.apache.org/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDIzODYyNQ_size_16_color_FFFFFF_t_70]: /images/20221120/48a4025fe3e34a239aea1eabd18e031d.png
[20201229234557504.png]: /images/20221120/c7a4be8dcc5f44dc9b8b9fbc88765afe.png
[2020122923470077.png]: https://img-blog.csdnimg.cn/2020122923470077.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDIzODYyNQ_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20201229235104990.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDIzODYyNQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDIzODYyNQ_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20201229235133402.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDIzODYyNQ==,size_16,color_FFFFFF,t_70
[20201229235618768.png]: https://img-blog.csdnimg.cn/20201229235618768.png
[20201229235653443.png]: https://img-blog.csdnimg.cn/20201229235653443.png
[20201230000230704.png]: https://img-blog.csdnimg.cn/20201230000230704.png
[20201230000252613.png]: https://img-blog.csdnimg.cn/20201230000252613.png