Logstash 設計了自己的 DSL —— 有點像 Puppet 的 DSL，或許因為都是用 Ruby 語言寫的吧 —— 包括有區域，註釋，資料型別(布林值，字串，數值，陣列，雜湊)，條件判斷，欄位引用等。

一、基本語法組成

logstash.conf配置檔案裡至少需要有input和output兩個部分構成

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

1、input配置

1.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.2、codec（定義編碼型別）
優化建議：直接輸入預定義好的 JSON 資料，這樣就可以省略掉 filter/grok 配置，從而減輕過濾器 logstash 的 CPU 負載消耗；具體操作如下：
1.2.1、修改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.2、重啟 nginx 應用，然後修改input/file 區段配置成下面這樣

input {

    file {

        path => "/var/log/nginx/access.log_json""

        codec => "json"

    }

}

2、filter過濾器配置

2.1、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.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.3、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"]

    }

}