Logstash收集多台设备的beats日志信息，如何添加多台源设备的源ip字段,并输出到kafka

我有需求是区分源ip地址的数据，基于Beats的配置好像不行，我的环境是filebeat --> Logstash --> kafka，想通过Logstash的geoip或者add_field增加ip地址信息。后来发现geoip无法识别内网(局域网)地址，如127.0.0.1和192.168.1.1等等，而add_field增加%{host}也不行，依然是主机名。。
各位大神  如何区分源（内网地址）ip信息，如何输出到kafka中呢？