一、filebeat

Filebeat，轻量级的开源日志文件数据搜集器。通常在需要采集数据的客户端安装 Filebeat，并指定目录与日志格式，Filebeat 就能快速收集数据，并发送给 logstash 进行解析，或是直接发给 Elasticsearch 存储，性能上相比运行于 JVM 上的 logstash 优势明显，是对它的替代。常应用于 EFLK 架构当中。

filebeat 结合 logstash 带来好处

1）通过 Logstash 具有基于磁盘的自适应缓冲系统，该系统将吸收传入的吞吐量，从而减轻 Elasticsearch 持续写入数据的压力

2）从其他数据源（例如数据库，S3对象存储或消息传递队列）中提取

3）将数据发送到多个目的地，例如S3，HDFS（Hadoop分布式文件系统）或写入文件

4）使用条件数据流逻辑组成更复杂的处理管道

二、ELFK

1.原理简介

在ELK中，Logstash 负责收集日志，将日志格式化并输出到 Elasticsearch 群集中；Elasticsearch 对格式化后的数据进行索引和存储；Kibana 从 ES 群集中查询数据生成图表，并进行前端数据的展示。

而在ELFK中，filebeat替代logstash收集日志，而logstash负责格式化日志和输出到es集群。

在这里插入图片描述

三、部署Filebeat+ELK

实验准备

Node1节点（2C/4G）：node1/192.168.17.50 Elasticsearch
Node2节点（2C/4G）：node2/192.168.17.51 Elasticsearch
Apache节点：apache/192.168.17.53 Logstash Kibana Apache
Filebeat节点：filebeat/192.168.17.54 Filebeat

在 Node1 节点上操作

1.解压安装

#上传软件包 filebeat-6.7.2-linux-x86_64.tar.gz 到/opt目录
tar zxvf filebeat-6.7.2-linux-x86_64.tar.gz
mv filebeat-6.7.2-linux-x86_64/ /usr/local/filebeat

2.设置 filebeat 的主配置文件

cd /usr/local/filebeat


filebeat.inputs:
- type: log                         #指定 log 类型，从日志文件中读取消息
  enabled: true
  paths:
    - /var/log/messages            #指定监控的日志文件
    - /var/log/*.log
  tags: ["sys"]		             #设置索引标签
  fields:                       #可以使用 fields 配置选项设置一些参数字段添加到 output 中
    service_name: filebeat
    log_type: syslog
    from: 192.168.17.52

--------------Elasticsearch output-------------------
(全部注释掉)

----------------Logstash output---------------------
output.logstash:
  hosts: ["192.168.17.50:5044"]      #指定 logstash 的 IP 和端口

在这里插入图片描述

3.启动 filebeat

nohup ./filebeat -e -c filebeat.yml > filebeat.out &
#-e：输出到标准输出，禁用syslog/文件输出
#-c：指定配置文件
#nohup：在系统后台不挂断地运行命令，退出终端不会影响程序的运行

在这里插入图片描述

4．在 Logstash 组件所在节点上新建一个 Logstash 配置文件


cd /etc/logstash/conf.d

vim filebeat.conf
input {
    beats {
        port => "5044"
    }
}

#filebeat发送给logstash的日志内容会放到message字段里面，logstash使用grok插件正则匹配message字段内容进行字段分割
#Kibana自带grok的正则匹配的工具：http://<your kibana IP>:5601/app/kibana#/dev_tools/grokdebugger
# %{IPV6}|%{IPV4} 为 logstash 自带的 IP 常量


filter {
  grok {
    match => ["message", "(?<remote_addr>%{IPV6}|%{IPV4})[\s\-]+\[(?<logTime>.*)\]\s+\"(?<method>\S+)\s+(?<url_path>.+)\"\s+(?<rev_code>\d+) \d+ \"(?<req_addr>.+)\" \"(?<content>.*)\""]
  }
}

output {
    elasticsearch {
        hosts => ["192.168.80.10:9200","192.168.80.11:9200"]
        index => "%{[fields][service_name]}-%{+YYYY.MM.dd}"
    }
    stdout {
        codec => rubydebug
    }
}

#启动 logstash
logstash -f filebeat.conf