logstash詳解

介紹

Logstash：資料處理引擎，可以處理每秒幾萬條的日志；它支援動态的從各種資料源搜集資料，并對資料進行過濾、分析、豐富、統一格式等操作，然後存儲到 ES

官網位址：https://www.elastic.co/cn/products/logstash

如圖：

詳解：

下面我們将文本資料同步到ES來做例子，其中版本為：logstash-5.2.1，elasticsearch-6.2.2

analysis.conf 配置檔案内容為

input {
    file {
        type => "lualog"
        path=> [ "C:/JWD/logs/*" ] #同步檔案的IP
        start_position => beginning
        stat_interval => 1 #設定多長時間檢測檔案是否修改 預設是1s
        #tags => "test" #添加标簽
        #設定多長時間掃描目錄，發現新檔案
        discover_interval => 15
        # 設定監聽間隔 各字段含義（從左至右）分、時、天、月、年，全為*預設含義為每分鐘都更新
    }
    file {
        type => "lsdclog"
        path=> [ "C:/JWD/server/Debug/Logs/*" ]  #同步檔案的IP
        start_position => beginning
        stat_interval => 1 #設定多長時間檢測檔案是否修改 預設是1s
        #tags => "test" #添加标簽
        #設定多長時間掃描目錄，發現新檔案
        discover_interval => 15
        # 設定監聽間隔 各字段含義（從左至右）分、時、天、月、年，全為*預設含義為每分鐘都更新
    }
}
 
filter {
   json{
        source => "message"
    }
    date {
        match => ["addTime", "yyyy.MM.dd HH.mm.ss","UNIX_MS"] #addTime是我日志中的一個時間字段！
        target => "@timestamp"
    }
    ruby {
        code => "event.set('timestamp', event.get('@timestamp').time.localtime + 8*60*60)"
    }
    ruby {
        code => "event.set('@timestamp',event.get('timestamp'))"
    }
    mutate{
        remove_field => ["message","timestamp"]
    }
}
output {
     if [type] == "lualog"{
        elasticsearch {
            action => "index"
            hosts => ["127.0.0.1:9200"] 
            index => "ls_%{+YYYYMMdd}_lua_log"
            document_type => "ls_%{+YYYYMMdd}_lua_log"
        }
     }
     if [type] == "lsdclog"{
        elasticsearch {
            action => "index"
            hosts =>["127.0.0.1:9200"] 
            index => "ls_%{+YYYYMMdd}_lsdc_log"
            document_type => "ls_%{+YYYYMMdd}_lsdc_log"
        }
     }
     stdout {
       codec => json
     }
}

           

1、将C:/JWD/logs/ 目錄下的所有檔案同步到本機ES中，索引名為ls_%{+YYYYMMdd}_lua_log，{+YYYYMMdd}為當日的日期

2、将C:/JWD/server/Debug/Logs/ 目錄下的所有檔案同步到本機ES中，索引名為ls_%{+YYYYMMdd}_lsdc_log，{+YYYYMMdd}為當日的日期

3、filter表示隻過濾json資料

注意：

1、這邊有個logstash自帶的坑，就是logstas同步日志的時候，索引是統計8點到第二天8點影響，是以要改索引建立為淩晨建立時間，如何修改，請看以上配置

2、要同步多個路徑下的檔案的時候，可以在input中file的 path=> [ “C:/JWD/logs/", "C:/JWD/server/Debug/Logs/”] 配置多個路勁（input中各個參數的意義，請參考：https://yq.aliyun.com/articles/152043?spm=a2c4e.11153940.blogcont154341.23.792e51fekIVnjR）

3、output的時候，如果在hosts =>[“127.0.0.1:9200”,“10.40.0.205:9200”] 配置多個路徑的時候，不是在兩台服務上各備份一份資料，而是把一份資料存到兩台服務上，所有每台服務上的資料都是不完整的

啟動代碼：logstash -f analysis.conf （其中analysis.conf為配置檔案的檔案名，在bin目錄下）