在logstash中，配置完輸入資料流後，接下來就是對資料進行解析，最簡單的方法莫過於使用過濾器(filter)與grok的組合。

在grok中，支援以正則表示式的方式提取所需要的資訊，其中，正則表示式又分兩種，一種是內建的正則表示式（可以滿足我們大部分的要求），一種是自定義的正則表示式，形式分別如下：

# 內建的正則表示式方式，下面的寫法表示從輸入的訊息中提取IP欄位，並命名為sip
%{IP:sip}
# 自定義的正則表示式，開始與終止符分別為(?與?)，下面的寫法表示獲取除,以外的字元，並命名為log_type
(?<log_type>[^,]+?)

以一個具體的例子來說明grok用法，假定需要解析的訊息內容樣本如下：

日誌型別:僵屍網路日誌, 源IP:192.168.101.251, 源埠:63726, 目的IP:124.127.48.41, 目的埠:1390, 攻擊型別:異常流量, 嚴重級別:低, 系統動作:被記錄, URL:-\n

為了快速判斷我們的正則表示式是否正確，不妨進行線上測試，地址為(線上測試地址)[http://grokdebug.herokuapp.com/]，最後的結果為：

^日誌型別:(?<log_type>[^,]+?), 源IP:%{IP:sip}, 源埠:%{NUMBER:sport:int}, 目的IP:%{IP:dip}, 目的埠:%{NUMBER:dport:int
 
}, 攻擊型別:(?<att_type>[^,]+?), 嚴重級別:(?<slevel>[^,]{1,}?), 系統動作:(?<sys_act>[^,]{1,}?), URL:(?<url>.+)$

在上面的表示式中，為了提高正則表示式的解析效率，我們需要進行整行匹配，於是添加了正則表示式的開始與結尾字元“^$”，此外為了便於統計與分析，我們還需要對相關型別進行轉換（如將字串轉換為整數），例如我們所需要的埠為整數，表示式為%{NUMBER:dport:int}。需要注意的是，grok也就支援兩種資料型別轉換，分別為float與int。

完整的logstash配置檔案如下所示。

input {
    # 監聽syslog輸出埠
    syslog {
        port => "514"
    }
}

filter {
    grok {
        match => {
            "message" => [
                "^日誌型別:(?<log_type>[^,]+?), 源IP:%{IP:sip}, 源埠:%{NUMBER:sport:int}, 目的IP:%{IP:dip}, 目的埠:%{NUMBER:dport:int}, 攻擊型別:(?<att_type>[^,]+?), 嚴重級別:(?<slevel>[^,]{1,}?), 系統動作:(?<sys_act>[^,]{1,}?), URL:(?<url>.+)$",
                "^日誌型別:(?<log_type>[^,]+?), 應用型別:(?<p_type>[^,]+?), 使用者名稱/主機:%{IP:sport}, 上行流量\(KB\):%{NUMBER:sflow:int}, 下行流量\(KB\):%{NUMBER:dflow:int}, 總流量\(KB\):%{NUMBER:all_flow:int}$"
            ]
        }
    }   
}

output {
    stdout {
        codec => rubydebug
    }
    elasticsearch {
        #   elasticsearch的服務地址
        hosts => ["192.168.101.204"]
        index => "logstash-%{+YYYY.MM.dd}"
        ssl => false
    }
}

在操作的過程中，相關問題的解決辦法：

[1]: max file descriptors [4096] for elasticsearch process is too low, increase to at least [65536]
[2]: max virtual memory areas vm.max_map_count [65530] is too low, increase to at least [262144]

檔案數量限制的解決辦法：

#   設定虛擬記憶體的數量
sysctl -w vm.max_map_count=999999
#   設定程序可開啟的檔案數量
ulimit -n 999999

無可用連線的錯誤資訊：

[2017-08-08T02:32:08,035][ERROR][logstash.outputs.elasticsearch] 
Attempted to send a bulk request to elasticsearch, 
but no there are no living connections in the connection pool.
Perhaps Elasticsearch is unreachable or down?
{:error_message=>"No Available connections", :class=>"LogStash::Outputs::ElasticSearch::HttpClient::Pool::NoConnectionAvailableError", :will_retry_in_seconds=>32}

解決辦法：
1. 仔細檢查配置檔案的elasticsearch的服務地址；
2. 仔細檢查elasticsearch是否開放了跨域設定；