input 詳解參考之前的文章
https://blog.csdn.net/gekkoou/article/details/80986017

input 官方詳解
https://www.elastic.co/guide/en/logstash/current/input-plugins.html

filter 官方詳解
https://www.elastic.co/guide/en/logstash/current/filter-plugins.html

output 官方詳解
https://www.elastic.co/guide/en/logstash/current/output-plugins.html

貼上收集 apache

input {
    # access日誌
    file {
        type => "apache_access"
        tag => "apache_access"
        path => ["/var/log/apache/access.log"]
        start_position => beginning
    }
    # error日誌
    file {
        type => "apache_error"
        tag => "apache_error"
        path => ["/var/log/apache/error.log"
 
]
        start_position => beginning
    }
}

filter {
    # 根據 input 新增的 type 來區分, 實現同時讀取兩種日誌, 也可以用 tag 來區分 (例如 if [tag] in "apache_access")
    if [type] == "apache_access"{
        # 文字片段切分的方式來切分日誌事件
        # 推薦使用grokdebugger來寫匹配模式: http://grokdebug.herokuapp.com/
        # grok官方詳解: https://www.elastic.co/guide/en/logstash/current/plugins-filters-grok.html
 

        grok {
            match => { "message" => "%{COMBINEDAPACHELOG}"}
        }
        # data外掛可以用來轉換你的日誌記錄中的時間字串, 然後轉存到 @timestamp 欄位裡
        date {
            match => [ "timestamp" , "dd/MMM/YYYY:HH:mm:ss Z" ]
        }
        # 通過geoip能獲取到很多的資訊，包括經緯度，國家，城市，地區等資訊
        geoip {
            # 來源於apache日誌中的clientip
            source => "clientip"
        }
        # useragent外掛可以幫助我們過濾出瀏覽器版本、型號以及系統版本
        useragent {
            source => "agent"
            target => "useragent"
        }
    } else if [type] == "apache_error"{
        grok {
            match => { "message" => "\[(?<mytimestamp>%{DAY:day} %{MONTH:month} %{MONTHDAY} %{TIME} %{YEAR})\] \[%{WORD:module}:%{LOGLEVEL:loglevel}\] \[pid %{NUMBER:pid}:tid %{NUMBER:tid}\]( \(%{POSINT:proxy_errorcode}\)%{DATA:proxy_errormessage}:)?( \[client %{IPORHOST:client}:%{POSINT:clientport}\])? %{DATA:errorcode}: %{GREEDYDATA:message}" }
        }
        date {
            match => [ "mytimestamp" , "EEE MMM dd HH:mm:ss.SSSSSS yyyy" ]
        }
    }

    #轉換型別 (integer, float, integer_eu, float_eu, string, boolean)
    #mutate {
    #    convert => ["ctime", "integer"]
    #    convert => ["lat", "float"]
    #}

    #當某條日誌資訊符合if規則時
    #if [field_name] == "value" {
    #    #drop可以跳過某些不想統計的日誌資訊
    #    drop {}
    #}

    #create_at為時間戳時需要轉換為0時區(UTC), 然後放入@timestamp欄位裡
    #date {
    #   match => ["create_at", "yyyy-MM-dd HH:mm:ss,SSS", "UNIX"]
    #   #match => ["create_at", "UNIX"]
    #   target => "@timestamp"
    #   locale => "cn"
    #   #remove_field => 'create_at' #刪除欄位
    #}

    # 執行ruby程式碼
    #ruby {
    #    code => "event.set('timestamp', event.get('@timestamp').time.localtime + 8*60*60)"
    #}
    #ruby {
    #    code => "event.set('@timestamp',event.get('timestamp'))"
    #}
}

# 輸出外掛將資料傳送到一個特定的目的地, 除了elasticsearch還有好多可輸出的地方, 例如file, csv, mongodb, redis, syslog等
output {
    if [type] == "apache_access"{
        elasticsearch {
            hosts => [ "localhost:9200" ]
            # 記錄的index索引名稱格式
            index => "apache-access-log-%{+YYYY.MM}"
        }
    } else if [type] == "apache_error"{
        elasticsearch {
            hosts => [ "localhost:9200" ]
            index => "apache-error-log"
        }
    }
}