統一日誌ELK部署配置（1）——filebeat

1、總體架構

2、filebeat介紹
fielbeat是基於logstash-forwarder的源碼改造而來，換句話說：filebeat就是最新版的logstash-forwarder。他負責從當前服務器獲取日誌而後轉發給Logstash或Elasticserach進行處理。
Filebeat是一個日誌文件託運工具，作爲一個agent安裝到服務器上，filebeat會監控日誌目錄或者指定的日誌文件，追蹤讀取這些文件（追蹤文件的變化，不停的讀），而且轉發這些信息到elasticsearch或者logstarsh中存放。
Filebeat包含兩個主要的組件，prospectors（探測器）和harvesters（收割機）。
prospectors負責管理全部的harvesters，和發現全部的須要讀取的日誌源。
若是輸入源是log（日誌），prospectors根據配置路徑，查找驅動器上的全部匹配的日誌文件，併爲每一個文件啓動一個收割機。
harvesters負責一行一行地讀取日誌文件，並將內容發送到指定的輸出。
Filebeat目前支持兩種prospector類型，log和stdin，每種類型均可以被定義屢次。

3、filebeat特色
一、部署簡單：二進制包，無其餘依賴，直接下週使用；
二、配置簡單：基於YAML，配置簡單，格式明瞭；
三、輕量級：定位於docker和虛擬機的日誌收集，運行佔用資源在10M左右；單核CPU的消耗；
四、性能高：性能比logstash高不少；
五、記錄收集位置：能記錄收集的內容位置，重啓後能基於在原基礎上收集；
六、支持多行合併：保證java異常堆棧記錄的完整性；
七、不支持事務；
八、功能單一：基於文件內容採集；能寫入kafka、redis、es，可是不支持從其讀；

4、filebeat安裝
下載安裝很是的簡單：
一、rpm -vi filebeat-5.6.4-x86_64.rpm
二、下載tar包，tar -zxvf filebeat-5.6.4-x86_64.tar.gz
而後經過./filebeat -c filebeat.yml 命令啓動

5、filebeat配置
#=========================== Filebeat prospectors =============================

filebeat.prospectors:

• input_type: log
  paths:

  • '/mnt/iss/service/order/nlogs/.log' #==監聽日誌文件全路徑 所有監聽用，不然直接寫具體文件名，也可模糊匹配。多個目錄時增長一行配置便可

  encoding: utf-8 #==編碼格式
  exclude_files: [".gz$ | .gc."] #==排除監聽的文件

  fields:
  serverName: 'user-service-006' ##==額外添加的字段，用於區分服務器名稱
  type: service-log ##==多種服務日誌時的區分字段
  fields_under_root: true ##==是否直接添加這些字段到日誌內容中
  scan_frequency: 3s ##==掃描文件的頻率
  #===== Multiline options
  multiline: ##==多行日誌的合併配置，用於異常堆棧內容時的處理
  pattern: '^{"date":'
  negate: true
  match: after
  timeout: 2s
  backoff: 1s
  max_backoff: 3s

  close_renamed: false ##文件重命名後是否中止監聽
  close_removed: true ##文件被刪除後是否中止監聽
  tail_files: true ## 是否從文件末尾讀取（啓動時）
  enabled: true
  filebeat.spool_size: 2048 ## 事件發送的閥值，超過閥值，強制刷新網絡鏈接
  filebeat.idle_timeout: 2s ## 事件發送的超時時間，即便沒有超過閥值，也會強制刷新網絡鏈接

  ignore_older: 24h ##日誌文件監聽超時時間閥值
  #================================ Outputs =====================================

#-------------------------- kafka output ------------------------------
output.kafka:
#== initial brokers for reading cluster metadata
hosts: ["kafka1:9092", "kafka2:9092", "kafka3:9092"]

#== message topic selection + partitioning
topic: 'elk-service-log'
partition.round_robin:
reachable_only: false

required_acks: 1
compressmax_message_bytes: 1000000
compression: snappy
timeout: 5m
keep_alive: 30s

logging.level: info