Pipeline分為三部分

• Input 資料輸入
• Filter 資料過濾整理增加
• Output 資料輸出

流程方向 Input→Filter→Output

Input

常見的套件如file、syslog、kafka、redis、beat 套件列表

這邊使用redis當隊列緩存

input {
    redis {
        data_type => "list" #list, channel, pattern_channel分三讀取的方式
        key => "dev" #redis list key
        host => "10.102.0.16" #redis host
        port => "6379" #redis prot
        db => 0 #redis db
        id => "elk-redis-dev-1" #當有兩個Logstash時識別用
    }
}

Filter

常見的套件如grok、mutate、drop、clone、geoip 套件列表

filter {
    #這邊使用Json套件把message欄位的資料轉成Json到content欄位
    json {
        source => "message"
        target => "content"
    }

    #這邊判斷是否有client_ip如果有的話透過geoip套件把ip轉成座標
    if [content][client_ip] {
        geoip {
            source => "[content][client_ip]"
            target => "geoip"
        }
    }

    #這個_jsonparsefailure tag是當轉json失敗時才有的處理
    if "_jsonparsefailure" in [tags] {
        #使用mutate套件增加欄位
        mutate {
            add_field => {
                "[content][level]" => "WARN"
                "[content][severity]" => 400
                "[content][message]" => "%{[message]}"
            }
        }
    }
}

Output

常見的套件如elastcisearch、file、kafka、statsd 套件列表

output {
    elasticsearch {
        #使用k8s的namespace和container name加日期組成索引名稱
        index => "%{[kubernetes][namespace]}-%{[kubernetes][container][name]}-@-%{+YYYY.MM.dd}"
        #elasticsearch會對列表裡的機器做load balance
        hosts => ["10.111.0.17:9200, 10.111.0.18:9200]
    }
}