日誌收集

一、整體架構

二、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，每種類型都可以被定義多次。

三、filebeat特點
1、部署簡單：二進制包，無其他依賴，直接下周使用；
2、配置簡單：基於YAML，配置簡單，格式明了；
3、輕量級：定位於docker和虛擬機的日誌收集，運行占用資源在10M左右；單核CPU的消耗；

4、性能高：性能比logstash高很多；
5、記錄收集位置：能記錄收集的內容位置，重啟後能基於在原基礎上收集；
6、支持多行合並：保證java異常堆棧記錄的完整性；
7、不支持事務；
8、功能單一：基於文件內容采集；能寫入kafka、redis、es，但是不支持從其讀；

四、filebeat安裝
下載安裝非常的簡單：
1、rpm -vi filebeat-5.6.4-x86_64.rpm
2、下載tar包，tar -zxvf filebeat-5.6.4-x86_64.tar.gz
然後通過./filebeat -c filebeat.yml 命令啟動

五、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

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