一、背景：

1、环境信息

前段时间为公司做了一个日志系统，
日志数据流：filebeat——>kafka——>logstash——>elasticsearch——>kibana

centos 7

filebeat 7.4

logstash 7.4

elasticsearch 7.4

kibana 7.4

kafka_2.13-2.5.0.tgz

2、为什么需要做多个output，并输出到本地

因为这个日志收集系统丢数据！！！


而且丢了数据还没有相关日志输出，filebeat、kafka、logstash、ES，还包括系统的message日志都没有记录丢数据的日志，根本不知道是在哪个环节丢的！！！


于是我司英俊帅气的架构师就想到了在应用日志中自定义输出唯一的日志ID，保证每一条需要收集的日志ID都是独一无二不会重复的，将此ID在logstash环节落盘，看被丢弃的日志是否走到了logstash环节。


于是我就需要想办法在logstash环节将ID输出到本地，于是就找到了logstash多output+CSV插件。

二、Logstash的output配置

output {
    if [fields][logtype] =~ '^log(.*)_nginx_(.+)'{
        elasticsearch {
            hosts => ["192.168.25.107:9200"]
            index => "%{[fields][logtype]}_%{+YYYY.MM.dd}"
            template => "/etc/logstash/geoip-template.json"
            template_name => "nginx_geoip"
            template_overwrite => true
        }
    }
    else if [fields][logtype] =~ '^log(.*)_app_(.+)'{
        elasticsearch {
            hosts => ["192.168.25.107:9200"]
            index => "%{[fields][logtype]}__%{[服务名称]}_%{+YYYY.MM.dd}"
        }
    }
    else {
        elasticsearch {
            hosts => ["192.168.25.107:9200"]
            index => "%{[fields][logtype]}_%{+YYYY.MM.dd}"
        }
    }
}

output {
    if [fields][logtype] =~ '^log(.*)_app__dscloud'{
        csv {
            path => "/data/logstash/logs/logstash_log_id_%{+YYYYMMDD}.csv"
            fields => ["[单号]"]
        }
    }
}

三、参考文档

官方文档
https://www.cnblogs.com/wangzhuxing/p/9678578.html#_label4