logstash与filebeat收集日志

青旅半醒 2022-06-03 00:20 394阅读 0赞

#### 1、背景介绍 ####

日志收集，采用的是 ELK 框架，即 elasticsearch，logstash，kibana，另外还有 filebeat 组件，其中 filebeat 用于扫描日志文件，将日志发送到 logstash 服务，logstash 服务则完成将日志切分，发送到 elasticsearch 服务。

#### 2、filebeat的部署步骤(日志 -> logstash) ####

*  下载安装包，并解压。
 *  创建xxx.filebeat.yml文件，注意内容的格式，否则你会找半天的错误都可能找不出原因。
 *  启动服务

[root@hadoop1 opt] tar -xf  filebeat-5.2.2-linux-x86_64.tar.gz  ./
    
    # 这里根据实际的业务需要编写filebeat.yml文件
    [root@hadoop1 filebeat-5.2.2-linux-x86_64] vim  swordfish.filebeat.yml

# 文件的具体内容如下：
    filebeat.prospectors:
    - input_type: log
      multiline.timeout: 1s
      paths:
        - /opt/swordfish/target/swordfish-all-1.0-SNAPSHOT/logs/exec-server*.log
      multiline:
        pattern: '^\['
        negate: true
        match: after
    # exclude_lines: ["^\\[DEBUG\\]"]
    output.logstash:
      hosts: ["192.168.14.148:9999"]

配置解释

multiline：适用于日志中每一条日志占据多行的情况，比如各种语言的报错信息调用栈。这个配置的下面包含如下配置：
       pattern：多行日志开始的那一行匹配的pattern
       negate：是否需要对pattern条件转置使用，不翻转设为true，反转设置为false
       match：匹配pattern后，与前面（before）还是后面（after）的内容合并为一条日志
       max_lines：合并的最多行数（包含匹配pattern的那一行）
       timeout：到了timeout之后，即使没有匹配一个新的pattern（发生一个新的事件），     也把已经匹配的日志事件发送出去

启动指令-后台启动

[root@hadoop1 filebeat-5.2.2-linux-x86_64]$ nohup ./filebeat -e -c udp-filebeat.yml -d publish &

#### 3、logstash的部署(logstash -> elasticsearch) ####

*  下载安装包，并解压。
 *  创建swordfish-pipeline.conf文件，注意内容的格式，否则你会找半天的错误都可能找不出原因。
 *  启动服务

# 解压
    [root@hadoop1 opt]# tar -xf logstash-5.2.2.tar.gz ./
    [root@hadoop1 opt]# vim swordfish-pipeline.conf

# 内容格式要注意
    input {
      beats {
        port => "9999"
      }
    }
    
    filter {
      grok {
        match => { "message" => ["%{LOGLEVEL:logLevel}\]\[%{NUMBER:nanoTime:integer}\](?<V1>.*)\[jobId=(?<jobId>[\w\d\-_]+)\](?<V2>.*)", "%{LOGLEVEL:logLevel}\]\[%{NUMBER:nanoTime:integer}\](?<V1>.)(?<V2>.*)"] }
        add_field => { "nest_msg" => "[%{logLevel}]%{V1}%{V2}" }
        remove_field => [ "message", "V1", "V2" ]
      }
    
      if ![jobId] {
      drop {}
    }
    
    mutate {
      convert => { "nanoTime" => "integer" }
    }
    }
    
    output {
      stdout {
        codec => json
      }
      elasticsearch {
        hosts => ["192.168.14.148:9200"]
        codec => json
        index => "swordfish-logstash-%{+YYYY.MM.dd}"
      }
    }

启动指令-后台启动

[root@hadoop1 logstash-5.2.2]# nohup bin/logstash -f udp-pipeline.conf &

#### 4es的部署我们之前有介绍过，下面我们去es-head页面去看一下日志是否在es中存储。 ####

![SouthEast][]  
上面图片中我们发现日志是可以收集到的。

[SouthEast]: /images/20220603/a5108689b52842ad9a1dfe13bf1fe871.png