ELK + F 建構

Elasticsearch + Logstash + Kibana + FileBeat 四個組件的組合

Logstash 在 ELK 架構中，是負責把收到的純文字資料，做特定的規則處理，就可以變成指定的欄位。建立欄位的好處是可以方便搜尋，而且也能做到比全文檢索更好的分析，可說是欄位切的好，查詢沒煩惱。我個人認為 Logstash 中最精華的部分就屬 Grok Filter。本篇將簡單教學如何透過 Logstash Grok Filter 建立 Elasticsearch 欄位。

FileaBeat 收集log資料送往 Logstash
Logstash 過濾資料再送往 Elasticsearch
Elasticsearch 搜尋資料引擎 Kibana會來請求搜尋log資料
Kibana UI介面

啟動辦法：

先去下載ELK + F 檔案, 記住版本上的對應.

E：到bin 目錄底下, 直接啟動 -> ./elasticsearch
L：到bin 目錄底下, 需修改config 及啟動時帶入config位置. 直接啟動 -> ./logstash -f ../config/logstash.conf
K：到bin 目錄底下, 直接啟動 ->./kibana
F：到bin 目錄底下, 直接啟動 ->./filebeat

各項設定：

filebeat.yml

name: localhost
output:
  logstash:
    enabled: true
    hosts:
      - localhost:5044
    index: "localhost"
filebeat.inputs:
    - type: log
      paths:
        - /Users/xxxxx/Desktop/log/xxxSystem/xxx.log
      multiline:
        pattern: '^\['
        negate: true
        match: after
      tags: ["restapi"]
#開啟debug模式
logging.level: debug
logging.selectors: [publish]
logging.to_files: true
logging.files:
    name: filebeat-localhost

logstash.conf

input {
  beats {
    port => 5044
  }
}

filter {
    grok {
        match => {"message" => [
                            "\[(?<XXOrderNo>[\w\d]*),(?<orderNo>[\w\d]*)\] %{TIMESTAMP_ISO8601:logDate} \[(?<threadName>[\w\d\s-]+)\] %{LOGLEVEL:logLevel}(?<emptySpace>\s+)\[(?<className>[\w.]+)\] - %{GREEDYDATA:message}"
                        ]
        }
        overwrite => [ "message" ]
    }

    # customize timestamp
    date {
      timezone => "Asia/Taipei"
      match => ["logDate", "ISO8601"]
      target => "@timestamp"
    }

    if "restapi" in [tags]{
      mutate {
          add_field => { "c_file_name" => "restapi" }
      }
    }

    mutate {
        remove_tag => [ "beats_input_codec_plain_applied" ]
        remove_field => [ "emptySpace" ]
    }
}

output {

  stdout {
    codec => rubydebug
  }

  # Sending properly parsed log events to elasticsearch
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "logstash-%{c_file_name}-%{+YYYY.MM.dd}"
  }

}

// 可使用default setting

// 可使用default setting
// 亦可修改 memory 設定等

查看結果：

可至Kibana 介面查看log檔案是否傳至Elasticsearch

參考網站：

PreviousReflect 應用 NextSocket.IO

Last updated 1 year ago

name: localhost output: logstash: enabled: true hosts: - localhost:5044 index: "localhost" filebeat.inputs: - type: log paths: - /Users/xxxxx/Desktop/log/xxxSystem/xxx.log multiline: pattern: '^\[' negate: true match: after tags: ["restapi"] #開啟debug模式 logging.level: debug logging.selectors: [publish] logging.to_files: true logging.files: name: filebeat-localhost

input { beats { port => 5044 } } filter { grok { match => {"message" => [ "\[(?<XXOrderNo>[\w\d]*),(?<orderNo>[\w\d]*)\] %{TIMESTAMP_ISO8601:logDate} \[(?<threadName>[\w\d\s-]+)\] %{LOGLEVEL:logLevel}(?<emptySpace>\s+)\[(?<className>[\w.]+)\] - %{GREEDYDATA:message}" ] } overwrite => [ "message" ] } # customize timestamp date { timezone => "Asia/Taipei" match => ["logDate", "ISO8601"] target => "@timestamp" } if "restapi" in [tags]{ mutate { add_field => { "c_file_name" => "restapi" } } } mutate { remove_tag => [ "beats_input_codec_plain_applied" ] remove_field => [ "emptySpace" ] } } output { stdout { codec => rubydebug } # Sending properly parsed log events to elasticsearch elasticsearch { hosts => ["localhost:9200"] index => "logstash-%{c_file_name}-%{+YYYY.MM.dd}" } }