使用Filebeat采集日志结合logstash过滤出你想要的日志

Posted 2023-04-23

参考技术A

使用外国大佬的开源项目，基本不要改什么就可快速搭建一套单机版ELK用于练手。
注意 ：logstash已被我改造，如果以该项目构建ELK记得更改logstash.conf。
ELK项目github链接： https://github.com/deviantony/docker-elk

这里对es不做过多描述，主要针对filebeat和logstash讲解。

Filebeat是一个轻量级的托运人，用于转发和集中日志数据。Filebeat作为代理安装在服务器上，监视您指定的日志文件或位置，收集日志事件，并将它们转发到Elasticsearch或 Logstash进行索引。
Filebeat的工作原理：启动Filebeat时，它会启动一个或多个输入，这些输入将查找您为日志数据指定的位置。对于Filebeat找到的每个日志，Filebeat启动一个收集器。每个收集器为新内容读取单个日志，并将新日志数据发送到libbeat，libbeat聚合事件并将聚合数据发送到您为Filebeat配置的输出。
官方流程图如下:

Logstash是一个具有实时流水线功能的开源数据收集引擎。Logstash可以动态统一来自不同来源的数据，并将数据标准化为您选择的目的地。为各种高级下游分析和可视化用例清理和民主化所有数据。
Logstash的优势：

接着进入正文，先讲下我的需求。

ELK以搭好，由elasticsearch+logstash+kibana 组成，

本文的filebeat作用是采集特定目录下的日志，并将其发送出去，但是它只能采集并无法对数据进行筛选，这时候就用到logstash了，logstash拥有众多插件可提供过滤筛选功能，由于logstash本身是基于jdk的，所以占用内存较大，而filebeat相较下，占用的内存就不是很多了。有图有真相：

所以可采用如下的方案采集筛选日志：

参考配置，只列出我用到的，详情见官方文档：

注：6.0以上该filebeat.yml需要挂载到/usr/share/filebeat/filebeat.yml,另外还需要挂载/usr/share/filebeat/data/registry 文件，避免filebeat容器挂了后，新起的重复收集日志。

我用到的logstash并不是用来采集日志的，而是对日志进行匹配筛选，所以不要跟随项目启动，只需单独启动，暴露5044端口，能接收到filebeat发送日志即可，也就是说，它只是起到一个加工并转发给elasticsearch的作用而已。
配置参考：

备注：

上面我自定义的格式是：

测试项目以上传到github
地址： https://github.com/liaozihong/ELK-CollectionLogs

参考链接：
ELK 之Filebeat 结合Logstash
Grok Debugger
ELK logstash 配置语法(24th)
Filebeat官方文档
Logstash官方文档

filebeat日志采集

架构一： filebeat -> logstash1 -> redis -> logstash2 -> elastash(集群) -> kibana 这里就不写安装程序的步骤了相信大家都没有难度： （软件安装可自行设计） 230，安装filebeat, logstash1 ,elastash 232，安装logstash2, redis, elastash ,kibana 注意：filebeat文件很注重文件格式 1，配置filebeat文件： [[email protected] filebeat]# cat /etc/filebeat/filebeat.yml filebeat: prospectors: # - #每个日志文件的开始 # paths: #定义路径 # - /var/www/logs/access.log #绝对路径 # input_type: log #日志类型为log # document_type: api4-nginx-accesslog # 此名称要与logstash定义的名称相对应，logstash要使用此名称做type判断使用 - paths: - /opt/apps/huhu/logs/ase.log input_type: log document_type: "ase-ase-log" encoding: utf-8 tail_files: true #每次最后一行 multiline.pattern: ‘^\[‘ #分割符 multiline.negate: true multiline.match: after #最后合并 #tags: ["ase-ase"] - paths: #收集json格式日志 - /var/log/nginx/access.log input_type: log document_type: "nginx-access-log" tail_files: true json.keys_under_root: true json.overwrite_keys: true registry_file: /var/lib/filebeat/registry output: #输出到230 logstash: hosts: ["192.168.0.230:5044"] shipper: logging: to_files: true files: path: /tmp/mybeat 2.配置230:logstash-->input-redis [[email protected] conf.d]# pwd /etc/logstash/conf.d [[email protected] conf.d]# cat nginx-ase-input.conf input { beats { port => 5044 codec => "json" }} output { if [type] == "nginx-access-log" { redis { #nginx日志写到redis信息 data_type => "list" key => "nginx-accesslog" host => "192.168.0.232" port => "6379" db => "4" password => "123456" }} if [type] == "ase-ase-log" { redis { #写到els日志写到redis信息 data_type => "list" key => "ase-log" host => "192.168.0.232" port => "6379" db => "4" password => "123456" }} } 3.redis写到elstach里，232服务器配置:logstash-->output-->resid->elstash [[email protected] conf.d]# pwd /etc/logstash/conf.d [[email protected] conf.d]# cat nginx-ase-output.conf input { redis { type => "nginx-access-log" data_type => "list" key => "nginx-accesslog" host => "192.168.0.232" port => "6379" db => "4" password => "123456" codec => "json" } redis { type => "ase-ase-log" data_type => "list" key => "ase-log" host => "192.168.0.232" port => "6379" db => "4" password => "123456" } } output { if [type] == "nginx-access-log" { elasticsearch { hosts => ["192.168.0.232:9200"] index => "nginx-accesslog-%{+YYYY.MM.dd}" }} if [type] == "ase-ase-log" { elasticsearch { hosts => ["192.168.0.232:9200"] index => "ase-log-%{+YYYY.MM.dd}" }} } 4，在232上配置elsaticsearch--->kibana 在kibana上找到ELS的索引即可。 架构二： filebeat -> redis -> logstash --> elsasctic --> kibana #缺点filebeat写进redis有限制，占时还没找到多个写入。 1.feilebeat配置： [[email protected] yes_yml]# cat filebeat.yml filebeat: prospectors: # - #每个日志文件的开始 # paths: #定义路径 # - /var/www/logs/access.log #绝对路径 # input_type: log #日志类型为log # document_type: api4-nginx-accesslog # 此名称要与logstash定义的名称相对应，logstash要使用此名称做type判断使用 - paths: - /opt/apps/qpq/logs/qpq.log input_type: log document_type: "qpq-qpq-log" encoding: utf-8 tail_files: true multiline.pattern: ‘^\[‘ multiline.negate: true multiline.match: after #tags: ["qpq-qpq-log"] registry_file: /var/lib/filebeat/registry output: redis: host: "192.168.0.232" port: 6379 db: 3 password: "123456" timeout: 5 reconnect_interval: 1 index: "pqp-pqp-log" shipper: logging: to_files: true files: path: /tmp/mybeat 2.由232redis-->els--kibana [[email protected] yes_yml]# cat systemlog.conf input { redis { type => "qpq-qpq-log" data_type => "list" key => "qpq-pqp-log" host => "192.168.0.232" port => "6379" db => "3" password => "123456" }} output { if [type] == "qpq-qpq-log"{ elasticsearch { hosts => ["192.168.0.232:9200"] index => "qpq-qpq-log-%{+YYYY.MM.dd}" } } } 3.在232上配置elsaticsearch--->kibana 在kibana上找到ELS的索引即可