ELK日誌管理之——logstash配置語法

Logstash 設計了本身的 DSL —— 有點像 Puppet 的 DSL，或許由於都是用 Ruby 語言寫的吧 —— 包括有區域，註釋，數據類型(布爾值，字符串，數值，數組，哈希)，條件判斷，字段引用等。

1、基本語法組成

logstash.conf配置文件裏至少須要有input和output兩個部分構成

input {
    #輸入
}
filter {
    #過濾匹配
}
output {
    #輸出
}

一、input配置

1.一、file{}（文件讀取）
監聽文件變化，記錄一個叫 .sincedb 的數據庫文件來跟蹤被監聽的日誌文件的當前讀取位（也就是時間戳）

input {

    file {

        path => ["/var/log/access.log", "/var/log/message"]      #監聽文件路徑
        type => "system_log"                                     #定義事件類型
        start_position => "beginning"                             #檢查時間戳
    }

}

參數說明：
exclude ：排除掉不想被監聽的文件

stat_interval ：logstash 每隔多久檢查一次被監聽文件狀態（是否有更新），默認是 1 秒。

start_position ：logstash 默認是從結束位置開始讀取文件數據，也就是說 logstash 進程會以相似 tail -f 的形式運行。若是你是要導入原有數據，把這個設定改爲 「beginning」，logstash 進程就按時間戳記錄的地方開始讀取，若是沒有時間戳則從頭開始讀取，有點相似cat，可是讀到最後一行不會終止，而是繼續變成 tail -f。

1.二、codec（定義編碼類型）
優化建議：直接輸入預約義好的 JSON 數據，這樣就能夠省略掉 filter/grok 配置，從而減輕過濾器 logstash 的 CPU 負載消耗；具體操做以下：
1.2.一、修改nginx配置文件，添加以下

logformat json '{"@timestamp":"$time_iso8601",'

               '"@version":"1",'

               '"host":"$server_addr",'

               '"client":"$remote_addr",'

               '"size":$body_bytes_sent,'

               '"responsetime":$request_time,'      #$request_time沒有雙引號代表該值爲int類型
               '"domain":"$host",'

               '"url":"$uri",'

               '"status":"$status"}';

access_log /var/log/nginx/access.log_json json;

1.2.二、重啓 nginx 應用，而後修改input/file 區段配置成下面這樣

input {

    file {

        path => "/var/log/nginx/access.log_json"" codec => "json" } }

二、filter過濾器配置

2.一、data（時間處理）

用來轉換日誌記錄中的時間字符串，變成LogStash::Timestamp 對象，而後轉存到 @timestamp 字段裏。

注意：由於在稍後的 outputs/elasticsearch 中index經常使用的 %{+YYYY.MM.dd} 這種寫法必須讀取 @timestamp數據，因此必定不要直接刪掉這個字段保留本身的時間字段，而是應該用 filters/date 轉換後刪除本身的字段！至於elasticsearch 中index使用 %{+YYYY.MM.dd}這種寫法的緣由後面會說明。

filter {

    grok {

        match => ["message", "%{HTTPDATE:logdate}"]

    }

    date {

        match => ["logdate", "dd/MMM/yyyy:HH:mm:ss Z"]

    }

}

2.二、grok （正則匹配）

filter {

     grok {
         match => [ "message",  "\s+(?<status>\d+?)\s+" ]        #跟python的正則有點差異
     }
}

優化建議：若是把 「message」 裏全部的信息都 grok 到不一樣的字段了，數據實質上就至關因而重複存儲了兩份。因此能夠用 remove_field 參數來刪除掉 message 字段，或者用 overwrite 參數來重寫默認的 message 字段，只保留最重要的部分。

filter {

    grok {

        patterns_dir => "/path/to/your/own/patterns"

        match => {

            "message" => "%{SYSLOGBASE} %{DATA:message}"

        }

        overwrite => ["message"]

    }

}

filter {

    grok {

        match => ["message", "%{HTTPDATE:logdate}"]

      remove_field => ["logdate"]

    }

}

2.三、GeoIP （地址查詢歸類）
GeoIP 是最多見的免費 IP 地址歸類查詢庫，同時也有收費版能夠採購。GeoIP 庫能夠根據 IP 地址提供對應的地域信息，包括國別，省市，經緯度等，對於可視化地圖和區域統計很是有用。

filter {
        geoip {
            source => "clientip"
            database => "/etc/logstash/GeoLiteCity.dat"       #需去官網下載ip庫放到本地
        }
}



filter {

        geoip {

            source => "message"               #若是能聯網可查詢在線ip庫
        }
}

注：geoip 插件的 「source」 字段能夠是任一處理後的字段，好比 「clientip」，可是字段內容卻須要當心！geoip 庫內只存有公共網絡上的 IP 信息，查詢不到結果的，會直接返回 null，而 logstash 的 geoip 插件對 null 結果的處理是：不生成對應的 geoip.字段。
因此在測試時，若是使用了諸如 127.0.0.1, 172.16.0.1, 182.168.0.1, 10.0.0.1 等內網地址，會發現沒有對應輸出！
GeoIP 庫數據較多，若是不須要這麼多內容，能夠經過 fields 選項指定本身所須要的。下例爲所有可選內容

filter {

    geoip {

        fields => ["city_name", "continent_code", "country_code2", "country_code3", "country_name", "dma_code", "ip", "latitude", "longitude", "postal_code", "region_name", "timezone"]

    }

}