Filebeat日志收集簡單使用

1.簡略介紹

輕量型日志采集器，用於轉發和匯總日志與文件。

官網： https://www.elastic.co/cn/beats/filebeat

2.本文實現的功能

 

 

3.事先必備：

至少一台Kafka節點。

4.配置Log4j，自定義代碼中日志信息輸出格式以及文件名稱

<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="info" schema="Log4J-V2.0.xsd" monitorInterval="600">
    <Properties>
        //存放日志的文件夾名稱
        <Property name="LOG_HOME">logs</Property>
        //日志文件名稱
        <property name="FILE_NAME">collector</property>
        //日志格式
        //[%d{yyyy-MM-dd'T'HH:mm:ss.SSSZZ}] 日志輸入時間，東八區
        //[%level{length=5}]    日志級別，debug、info、warn、error
        //[%thread-%tid]    當前線程信息
        //[%logger] 當前日志信息所屬類全路徑
        //[%X{hostName}]    當前節點主機名。需要通過MDC來自定義。
        //[%X{ip}]  當前節點ip。需要通過MDC來自定義。
        //[%X{applicationName}] 當前應用程序名。需要通過MDC來自定義。
        //[%F,%L,%C,%M] %F：當前日志信息所屬的文件(類)名，%L：日志信息在所屬文件中的行號，%C：當前日志所屬文件的全類名，%M：當前日志所屬的方法名
        //[%m]  日志詳情
        //%ex   異常信息
        //%n    換行
        <property name="patternLayout">[%d{yyyy-MM-dd'T'HH:mm:ss.SSSZZ}] [%level{length=5}] [%thread-%tid] [%logger]
            [%X{hostName}] [%X{ip}] [%X{applicationName}] [%F,%L,%C,%M] [%m] ## '%ex'%n
        </property>
    </Properties>
    <Appenders>
        //日志輸出至控制台
        <Console name="CONSOLE" target="SYSTEM_OUT">
            <PatternLayout pattern="${patternLayout}"/>
        </Console>
        //全量日志信息
        <RollingRandomAccessFile name="appAppender" fileName="${LOG_HOME}/app-${FILE_NAME}.log"
                                 filePattern="${LOG_HOME}/app-${FILE_NAME}-%d{yyyy-MM-dd}-%i.log">
            <PatternLayout pattern="${patternLayout}"/>
            <Policies>
                <TimeBasedTriggeringPolicy interval="1"/>
                <SizeBasedTriggeringPolicy size="500MB"/>
            </Policies>
            <DefaultRolloverStrategy max="20"/>
        </RollingRandomAccessFile>
        //日志級別是warn以上的日志信息
        <RollingRandomAccessFile name="errorAppender" fileName="${LOG_HOME}/error-${FILE_NAME}.log"
                                 filePattern="${LOG_HOME}/error-${FILE_NAME}-%d{yyyy-MM-dd}-%i.log">
            <PatternLayout pattern="${patternLayout}"/>
            <Filters>
                <ThresholdFilter level="warn" onMatch="ACCEPT" onMismatch="DENY"/>
            </Filters>
            <Policies>
                <TimeBasedTriggeringPolicy interval="1"/>
                <SizeBasedTriggeringPolicy size="500MB"/>
            </Policies>
            <DefaultRolloverStrategy max="20"/>
        </RollingRandomAccessFile>
    </Appenders>
    <Loggers>
        <!-- 業務相關 異步logger -->
        <AsyncLogger name="com.sakura.*" level="info" includeLocation="true">
            <AppenderRef ref="appAppender"/>
        </AsyncLogger>
        <AsyncLogger name="com.sakura.*" level="info" includeLocation="true">
            <AppenderRef ref="errorAppender"/>
        </AsyncLogger>
        <Root level="info">
            <Appender-Ref ref="CONSOLE"/>
            <Appender-Ref ref="appAppender"/>
            <AppenderRef ref="errorAppender"/>
        </Root>
    </Loggers>
</Configuration>

5.Filebeat安裝

#上傳Filebeat至任意目錄下
cd /usr/local/software
tar -zxvf filebeat-6.6.0-linux-x86_64.tar.gz -C /usr/local/
cd /usr/local
mv filebeat-6.6.0-linux-x86_64/ filebeat-6.6.0
## 配置filebeat
vim /usr/local/filebeat-5.6.2/filebeat.yml
##可參考下方配置信息
啟動：
## 檢查配置是否正確
cd /usr/local/filebeat-6.6.0
./filebeat -c filebeat.yml -configtest
## Config OK
## 啟動filebeat
/usr/local/filebeat-6.6.0/filebeat &
#查看是否啟動成功
ps -ef | grep filebeat

Filebeat配置參考信息

###################### Filebeat Configuration Example #########################
filebeat.prospectors:

- input_type: log

  paths:
    ## app-服務名稱.log, 為什么寫死，防止發生輪轉抓取歷史數據
    - /usr/local/logs/app-collector.log        #日志文件地址
  #定義寫入 ES 時的 _type 值
  document_type: "app-log"
  multiline:
    #pattern: '^\s*(\d{4}|\d{2})\-(\d{2}|[a-zA-Z]{3})\-(\d{2}|\d{4})'   # 指定匹配的表達式（匹配以 2017-11-15 08:04:23:889 時間格式開頭的字符串）
    pattern: '^\['                              # 指定匹配的表達式（匹配以 "{ 開頭的字符串）。具體以哪種形式進行匹配要根據實際的日志格式來配置。
    negate: true                                # 是否必須匹配到
    match: after                                # 以[開頭的多行數據，從第二行開始合並到上一行的末尾
    max_lines: 2000                             # 最大的行數，多余的不再合並到上一行末尾
    timeout: 2s                                 # 如果在規定時間沒有新的日志事件就不等待后面的日志，提交數據
  fields:
    logbiz: collector
    logtopic: app-log-collector   ## 按服務划分用作kafka topic
    evn: dev


- input_type: log


  paths:
    - /usr/local/logs/error-collector.log
  document_type: "error-log"
  multiline:
    #pattern: '^\s*(\d{4}|\d{2})\-(\d{2}|[a-zA-Z]{3})\-(\d{2}|\d{4})'   # 指定匹配的表達式（匹配以 2017-11-15 08:04:23:889 時間格式開頭的字符串）
    pattern: '^\['                              # 指定匹配的表達式（匹配以 "{ 開頭的字符串）
    negate: true                                # 是否匹配到
    match: after                                # 合並到上一行的末尾
    max_lines: 2000                             # 最大的行數
    timeout: 2s                                 # 如果在規定時間沒有新的日志事件就不等待后面的日志
  fields:
    logbiz: collector
    logtopic: error-log-collector   ## 按服務划分用作kafka topic
    evn: dev
    
output.kafka:
  enabled: true
  hosts: ["192.168.204.139:9092"]
  topic: '%{[fields.logtopic]}'
  partition.hash:
    reachable_only: true
  compression: gzip
  max_message_bytes: 1000000
  required_acks: 1
logging.to_files: true

6.在kafka上創建對應的topic

略

7.啟動kafka、代碼程序，最后啟動Filebeat。

這個時候一切正常的話，Filebeat就會將數據推送至Kafka。可以進入到kafka的“kafka-logs/{topic-partition}”目錄下查看日志文件等，當對程序進行訪問時相應的日志信息將會被Filebeat采集推送到Kafka指定的topic上。

8.使用Logstash消費Kafka中的數據

A.安裝Logstash

Logstash安裝及基礎命令：https://www.cnblogs.com/monument/p/12950290.html

B.配置Logstash啟動腳本

input {
  kafka {
    ## app-log-服務名稱
    topics_pattern => "app-log-.*"
    bootstrap_servers => "192.168.11.51:9092"
    codec => json
    consumer_threads => 1    ## 因為只設置了一個partition，所以消費者線程數設置為1
    decorate_events => true
    #auto_offset_rest => "latest"
    group_id => "app-log-group"
   }
   kafka {
    ## error-log-服務名稱
    topics_pattern => "error-log-.*"
    bootstrap_servers => "192.168.11.51:9092"
    codec => json
    consumer_threads => 1
    decorate_events => true
    #auto_offset_rest => "latest"
    group_id => "error-log-group"
   }
   
}

filter {
  
  ## 時區轉換
  ruby {
    code => "event.set('index_time',event.timestamp.time.localtime.strftime('%Y.%m.%d'))"
  }

  if "app-log" in [fields][logtopic]{
    grok {
        ## 表達式
        match => ["message", "\[%{NOTSPACE:currentDateTime}\] \[%{NOTSPACE:level}\] \[%{NOTSPACE:thread-id}\] \[%{NOTSPACE:class}\] \[%{DATA:hostName}\] \[%{DATA:ip}\] \[%{DATA:applicationName}\] \[%{DATA:location}\] \[%{DATA:messageInfo}\] ## (\'\'|%{QUOTEDSTRING:throwable})"]
    }
  }

  if "error-log" in [fields][logtopic]{
    grok {
        ## 表達式
        match => ["message", "\[%{NOTSPACE:currentDateTime}\] \[%{NOTSPACE:level}\] \[%{NOTSPACE:thread-id}\] \[%{NOTSPACE:class}\] \[%{DATA:hostName}\] \[%{DATA:ip}\] \[%{DATA:applicationName}\] \[%{DATA:location}\] \[%{DATA:messageInfo}\] ## (\'\'|%{QUOTEDSTRING:throwable})"]
    }
  }
  
}

## 測試輸出到控制台：
output {
  stdout { codec => rubydebug }
}

## elasticsearch，未實現：
output {

  if "app-log" in [fields][logtopic]{
    ## es插件
    elasticsearch {
          # es服務地址
        hosts => ["192.168.11.35:9200"]
        # 用戶名密碼      
        user => "elastic"
        password => "123456"
        ## 索引名，+ 號開頭的，就會自動認為后面是時間格式：
        ## javalog-app-service-2019.01.23 
        index => "app-log-%{[fields][logbiz]}-%{index_time}"
        # 是否嗅探集群ip：一般設置true；http://192.168.11.35:9200/_nodes/http?pretty
        # 通過嗅探機制進行es集群負載均衡發日志消息
        sniffing => true
        # logstash默認自帶一個mapping模板，進行模板覆蓋
        template_overwrite => true
    } 
  }
  
  if "error-log" in [fields][logtopic]{
    elasticsearch {
        hosts => ["192.168.11.35:9200"]    
        user => "elastic"
        password => "123456"
        index => "error-log-%{[fields][logbiz]}-%{index_time}"
        sniffing => true
        template_overwrite => true
    } 
  }
  

}

C.啟動Logstash

過程較慢，CPU、內存占用極高。啟動完成后可以在控制台（上一步配置的是將消息輸出到控制台）看到消息輸出。

9.將Logstash消費的數據推送到ElasticSearch

待續（太耗資源了，機器內存有限，裝不了這么多節點，待擴容、整理后再寫。）。